/*******************************************************************************
 * Copyright (c) 2005, 2009 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 API and Implementation
 *    Kevin O'Leary, Intel - Implementation
 *    IBM - Portions of code moved from older RAC (by Andy Kaylor)
 *
 * $Id: RACmdHandlers.c,v 1.62 2009/04/14 20:33:28 jwest Exp $
 *
 *******************************************************************************/ 

#include "tptp/TPTPTypes.h"
#include "tptp/TPTPSupportTypes.h"
#include "tptp/TransportSupport.h"
#include "tptp/TPTPBaseTL.h"
#include "tptp/BaseTLLog.h"
#include "tptp/ProcCtlUtils.h"
#include "tptp/compatibility/CmdConv.h"
#include "ClientCTL.h"
#include "RACClientSupport.h"
#include "RACmdHandlers.h"
#include "tptp/ACFlags.h"
#include "Connect2AC.h"
#include "user.h"

/* 
 * This file provides the code to handle commands that are sent from the client
 *   to the RAC.  The handlers here emulate the behavior of the RAC in their
 *   communications with the client while translating these commands into the
 *   necessary equivalents to provide the same functionality using the
 *   new Agent Controller.
 */

static int sendFileServerPort (ra_command_t *command, client_connection_block_t* ccb);

/* TODO: Begin -- functions looking for a home. */


/** SEND_ERROR_MESSAGE  *******************************************************************
  */
static void sendErrorMessage(const char *messageId,
							 const char *message,
							 ra_uint_t  contextId,
							 client_connection_block_t *ccb) {

	ra_message_t *errorMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
	ra_command_t *errorCommand=ra_addCommandToMessage(errorMessage, NULL);
	errorCommand->tag=RA_ERROR_STRING;
	errorCommand->info.error_string.context=contextId;
	/* ##TO_DO:  add proper processId */
	errorCommand->info.error_string.processId=0;
	ra_createRASTRING(&errorCommand->info.error_string.agent, "Agent Contoller");
	ra_createRASTRING(&errorCommand->info.error_string.messageId, messageId);
	ra_createRASTRING(&errorCommand->info.error_string.message, message);
	ra_forwardMessage(errorMessage, ccb);
	ra_destroyMessage(errorMessage, TRUE);
	return;
}

/** SCRUB_ALL_PROCESSES  ***************************************************************
  * This function tests each of the processes known by the server to determine if they
  * have exited yet.  This functionality is substantaily different on each platform.  In
  * the case of Windows, it walks the process list trying to open each process.  In the
  * case of Linux, it opens /proc and then looks for the process in the process list.
  */
static void scrubAllProcesses(process_list_t* list) {
/* Bug 59316 begins */
	process_list_node_t *current;
	PID pid, pid2;
	tl_state_data_t*  stateData = (tl_state_data_t*) list->stateData;

	ra_mutexEnter(&list->lock); /* Bug 65166 */
	current = list->head;
	while(current != NULL) {
		process_list_node_t *next=current->next;
		pid = current->process->pid;
#if defined __linux__
		pid2 = current->process->pid2;
#else
		pid2 = pid;
#endif
		if(!isProcessActive(pid) && !isProcessActive(pid2)) { /* Bug 62160 */
//			ra_logServiceMessage(__FILE__, __LINE__,RA_INFORMATION, "Process %d does not exist so scrub it", pid);
			scrubProcess(list, current->process, stateData);
		}
		current = next;
	}
	ra_mutexExit(&list->lock); /* Bug 65166 */

	return;
/* Bug 59316 ends */
}


static process_t* findProcess(cctl_state_data_t* cctlData, PID pid) {
	if(!cctlData->isPolling) { /* Bug 59316 */
		scrubAllProcesses(cctlData->processList);
	}
	return ra_processFind(cctlData->processList, pid);
}


/*
 * The thread function of the polling mechanism
 */
#ifdef _WIN32
DWORD WINAPI processPollingThread(LPVOID args) {
#else
void *processPollingThread(void *args) {
#endif
	cctl_state_data_t* cctlData = (cctl_state_data_t*)args;
	while(TRUE) {
		scrubAllProcesses(cctlData->processList);
		SLEEP(5000); /* 5 sec */
	}
	return 0;
}

/* TODO: End -- functions looking for a home. */

/* NOTE: All commands will be released when we return from these functions.
 *       If you need a copy, make a copy using ra_cloneCommand
 *
 */

BOOL handleAuthenticate( client_connection_block_t* ccb, ra_command_t* command )
{
	ra_uint_t tag = command->tag;
	cctl_state_data_t*  cctlData;

	cctlData = (cctl_state_data_t*)ccb->stateData->implData;
	if(tag == RA_AUTHENTICATE) {
		ra_message_t *replyMessage; /* reply */
		ra_command_t *replyCommand; /* reply */
		BOOL success = FALSE;
		ra_string_t user = command->info.authenticate.user;
		ra_string_t password = command->info.authenticate.passwd;

		TPTP_LOG_INFO_MSG1(ccb->stateData, "Trying to authenticate user: %s", user.data);

		if((user.length > 0) && tptpUserAllowed(ccb->stateData, user.data, cctlData->user, cctlData->userlist)) { /* Bug 145071 */
			TPTP_LOG_DEBUG_MSG1(ccb->stateData, "User %s is allowed to use the Agent Controller", user.data);
			success = tptpAuthenticateUser(ccb->stateData, user.data, password.data);
		}
		else {
			TPTP_LOG_INFO_MSG1(ccb->stateData, "User %s is not allowed to use the Agent Controller", user.data);
			success = FALSE;
		}

		/* Formulate the reply to the client */
		replyMessage = ra_createMessage(RA_CONTROL_MESSAGE, 0);
		replyCommand = ra_addCommandToMessage(replyMessage, NULL);

		if(success) {
			ccb->authenticated = TRUE;
			TPTP_LOG_INFO_MSG1(ccb->stateData, "Authentication successful: %s", user.data);
			replyCommand->tag = RA_AUTHENTICATION_SUCCESSFUL;
			ra_createRASTRING(&replyCommand->info.authenticate_successful.key, "none");
		}
		else {
			ccb->authenticated = FALSE;
			TPTP_LOG_INFO_MSG1(ccb->stateData, "Authentication failed: %s", user.data);
			replyCommand->tag = RA_AUTHENTICATION_FAILED;
			replyCommand->info.authenticate_failed.ticket = 1;
		}

		ra_forwardMessage(replyMessage, ccb);
		ra_destroyMessage(replyMessage, FALSE);

		return TRUE; /* Command processed sucessfully */
	}
	else {
		return FALSE; /* Command cannot be processed */
	}
}

BOOL handleLaunchProcess( client_connection_block_t* ccb, ra_command_t* command )
{
	cctl_state_data_t*  cctlData;
	PID real_pid;
	ra_array_t finalEnvironment;

	SOCKET consoleSock; /* 9707 */
	ra_string_t uuid;

	cctlData = (cctl_state_data_t*)ccb->stateData->implData;
	if ( cctlData == NULL )
	{
		sendErrorMessage("RAC003",
						 "Memory allocation error",
						 command->info.launch_process.context,
						 ccb);
		return FALSE;
	}
	/* This does NOT map 1-to-1 to launching a process using the process controller
	 *   In addition to launching the process, we need to look for any agents that
	 *   the client expects to be associated with this process and attach to them
	 */
	
	//ra_logServiceMessage(__FILE__, __LINE__, RA_DEBUG, "Console publish 0x%x, desired 0x%x  socket=%d", command->info.launch_process.consoleIP, buffer, client->connection->handle.socket);

	real_pid=0;
	
	/* Launch Process */
	real_pid=startProcess(ccb, &command->info.launch_process.executable, 
							 &command->info.launch_process.arguments,
							 &command->info.launch_process.location,
							 &command->info.launch_process.environment,
							 0 /* This value isn't used. */,
							 (unsigned short)command->info.launch_process.consolePort,
							 &finalEnvironment,
							 &consoleSock,
							 &uuid); /* 9707 */


	/* Was the launch successful? Update the process table and
	   formulate a reply to the client*/
	if(real_pid != -1) {
		unsigned long i;
		ra_message_t *replyMessage;
		ra_command_t *replyCommand;
		process_t *process;

		/* Create the process information */
		process=ra_processCreate(cctlData->processList);
		ra_copyRASTRING(&process->uuid, &uuid);
		ra_destroyRASTRING(&uuid);
		process->pid=real_pid;
		process->console = consoleSock; /* 9707 */

		/* Place the application name in the process. */
		ra_copyRASTRING(&process->processName, &command->info.launch_process.executable);

		/* Save the agents this client is interested in */
		for(i=0; i<command->info.launch_process.agents.length; i++) {
			agent_t *agent;

			/* Initialize agent */
			agent=ra_agentCreate(process, FALSE);
			ra_copyRASTRING(&agent->agentName, (ra_string_t*)(command->info.launch_process.agents.data[i]));
			agent->attached=TRUE;
			agent->client=(client_t*)tptp_malloc(sizeof(client_t));
			if (agent->client != NULL ) {
				agent->client->clientID = ccb->clientID;
			}
			agent->context=command->info.launch_process.context;
			//ra_logServiceMessage(__FILE__, __LINE__, RA_INFORMATION, "Adding agent interest: %s",agent->agentName.data);
		}

		/* Formulate the reply to the client */
		replyMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
		replyCommand=ra_addCommandToMessage(replyMessage, NULL);


		/* Set the context */
		replyCommand->info.process_launched.context=command->info.launch_process.context;
		replyCommand->tag=RA_PROCESS_LAUNCHED;
		replyCommand->info.process_launched.processId=real_pid;
		ra_copyRASTRING(&replyCommand->info.process_launched.arguments, &command->info.launch_process.arguments);
		ra_copyRASTRING(&replyCommand->info.process_launched.executable, &command->info.launch_process.executable);
		ra_copyRASTRING(&replyCommand->info.process_launched.processUUID, &process->uuid);
		replyCommand->info.process_launched.environment.length=finalEnvironment.length;
		replyCommand->info.process_launched.environment.data=finalEnvironment.data;

		/* Send a message back to the client informing the process was launched */
		ra_forwardMessage(replyMessage, ccb);

		/* Free used memory */
		ra_destroyMessage(replyMessage, TRUE);
	}
	else {
		sendErrorMessage("RAC002",
						 "Process Launch Failed",
						 command->info.launch_process.context,
						 ccb);
	}

	return TRUE;
}


BOOL handleKillProcess( client_connection_block_t* ccb, ra_command_t* command )
{
	process_t*          process;
	cctl_state_data_t*  cctlData;
	ra_message_t *killMessage;
	ra_command_t *killCommand;

	PID pid = command->info.kill_process.processId;

	cctlData = (cctl_state_data_t*)ccb->stateData->implData;
	if ( cctlData == NULL )
	{
		sendErrorMessage("RAC003",
						 "Memory allocation error",
						 command->info.kill_process.context,
						 ccb);
		return FALSE;
	}

	/* Find the process */
	process = findProcess(cctlData, command->info.kill_process.processId);

	/* If no such process return an error */
	if ( !process ) { /* 215066 */
		/* TO_DO:  Set proper error number */
		sendErrorMessage("RAC007",
						 "No such agent",
						 command->info.attach.context,
						 ccb);
		return FALSE;
	}
	if(process != NULL) { /* Process found */
		agent_list_t *agents = &(process->active_agents);

		if(agents) {
			agent_list_node_t *agentNode = agents->head;

			if (agentNode) {
				killMessage = ra_createMessage(RA_CONTROL_MESSAGE, 0);
				killCommand = ra_addCommandToMessage(killMessage, NULL);
				killCommand->tag = RA_CUSTOM_COMMAND;
				killCommand->info.custom_command.processId = pid;
				ra_createRASTRING(&killCommand->info.custom_command.message, "KILLED");

				while (agentNode) {
					agent_t* agent = agentNode->agent;

					killCommand->info.custom_command.context = agent->context;
					ra_copyRASTRING(&killCommand->info.custom_command.agent, &agent->agentName);
					ra_forwardMessage(killMessage, ccb);

					//We need to stop the agent from flushing.
					if (agent->IPCBufCreated == TRUE) {
						stopAgentDataFlush(ccb->stateData, agentNode->agent);
					}
					agentNode = agentNode -> next;
				}
				ra_destroyMessage(killMessage, TRUE);
			}
		}
	}
	/* Bug 64712 */

	if(killProcess(ccb, command->info.kill_process.processId)) {

		if(!cctlData->isPolling) { /* Bug 65166 */
			ra_mutexEnter(&cctlData->processList->lock); /* Bug 65166 */
			if(process) {
				scrubProcess(cctlData->processList, process, ccb->stateData);
			}
			ra_mutexExit(&cctlData->processList->lock); /* Bug 65166 */
		}
	}


	return TRUE;
}

BOOL handleQueryProcessList( client_connection_block_t* ccb, ra_command_t* command )
{

	cctl_state_data_t*  cctlData;
	ra_message_t *replyMessage;
	ra_command_t *replyCommand;

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

	if(!cctlData->isPolling) { /* Bug 59316 */
		scrubAllProcesses(cctlData->processList);
	}

	replyMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
	replyCommand=ra_addCommandToMessage(replyMessage, NULL);

	replyCommand->tag=RA_PROCESS_LIST;
	replyCommand->info.registered_process_list.context=command->info.query_process_list.context;
	replyCommand->info.registered_process_list.processes.length=cctlData->processList->process_count;

	/* Send all the processes */
	if(replyCommand->info.registered_process_list.processes.length) {
		process_list_node_t *current;
		unsigned int i;
		replyCommand->info.registered_process_list.processes.data=(void **)tptp_malloc(sizeof(tptp_uint32*)*replyCommand->info.registered_process_list.processes.length);
		current=cctlData->processList->head;
		for(i=0; i<replyCommand->info.registered_process_list.processes.length; i++) {
			((tptp_uint32**)replyCommand->info.registered_process_list.processes.data)[i]=(tptp_uint32*)tptp_malloc(sizeof(tptp_uint32));
			*(((tptp_uint32**)replyCommand->info.registered_process_list.processes.data)[i])=current->process->pid;
			current=current->next;
		}
	}

	/* Forward the message */
	ra_forwardMessage(replyMessage, ccb);
	ra_destroyMessage(replyMessage, TRUE);

	return TRUE;
}

BOOL handleQueryAgentList( client_connection_block_t* ccb, ra_command_t* command )
{
	ra_message_t *replyMessage;
	ra_command_t *replyCommand;
	process_t *process;
	cctl_state_data_t*  cctlData;

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

	replyMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
	replyCommand=ra_addCommandToMessage(replyMessage, NULL);
	replyCommand->tag=RA_AGENT_LIST;
	replyCommand->info.registered_agents_list.context=command->info.query_agent_list.context;
	replyCommand->info.registered_agents_list.processId=command->info.query_agent_list.processId;


/*
#if !defined _AIX && !defined MVS
	if(ra_getProcessNameByPID(replyCommand->info.registered_agents_list.processId,
						   namebuffer,
						   sizeof(namebuffer),
						   &nameSizeNeeded)) {
		ra_createRASTRING(&replyCommand->info.registered_agents_list.executable, namebuffer);
	}
	else {
		ra_createRASTRING(&replyCommand->info.registered_agents_list.executable, "unknown");
	}
#else
	*/
	ra_createRASTRING(&replyCommand->info.registered_agents_list.executable, "unknown");
//#endif

	/* Find the process */
	process = findProcess(cctlData, command->info.query_agent_list.processId);
	/* Is there such a process */
	if(process) {
		replyCommand->info.registered_agents_list.agents.length=process->active_agent_count;
		if(replyCommand->info.registered_agents_list.agents.length) {
			agent_list_node_t *agentNode;
			unsigned int i;
			replyCommand->info.registered_agents_list.agents.data=(void**)tptp_malloc(sizeof(ra_string_t*)*replyCommand->info.registered_agents_list.agents.length);
			agentNode=process->active_agents.head;
			i=0;
			while(agentNode!=NULL && i<replyCommand->info.registered_agents_list.agents.length) {
				((ra_string_t**)replyCommand->info.registered_agents_list.agents.data)[i]=(ra_string_t*)tptp_malloc(sizeof(ra_string_t));
    			ra_copyRASTRING(((ra_string_t**)replyCommand->info.registered_agents_list.agents.data)[i], &agentNode->agent->agentName);
				i++;
				agentNode=agentNode->next;
			}
		}
	}
	else {
		replyCommand->info.registered_agents_list.agents.length=0;
	}

	ra_forwardMessage(replyMessage, ccb);
	ra_destroyMessage(replyMessage, TRUE);

	return TRUE;
}

BOOL handleQueryAgentState( client_connection_block_t* ccb, ra_command_t* command )
{
	ra_message_t *msg;
	ra_command_t *cmd;
	agent_t *agent = NULL;
	process_t *process;
	tptp_int32 status;
	cctl_state_data_t*  cctlData;

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

	msg = ra_createMessage(RA_CONTROL_MESSAGE, 0);
	cmd = ra_addCommandToMessage(msg, NULL);

	cmd->info.agent_state.context = command->info.agent_query_state.context;
	cmd->info.agent_state.processId = command->info.agent_query_state.processId;

	status =  findProcessAndAgent(command->info.agent_query_state.processId, 
			&command->info.agent_query_state.agent, TRUE, cctlData->processList, &agent, &process);
	if(agent != NULL) {
		ra_copyRASTRING(&cmd->info.agent_state.agent, &command->info.agent_query_state.agent);
		if(agent->attached) {
			cmd->tag = RA_AGENT_ATTACHED;
		}
		else {
			cmd->tag = RA_AGENT_DETACHED;
		}
		ra_forwardMessage(msg, ccb);
		ra_destroyMessage(msg, TRUE);
	}

	return TRUE;
}

BOOL handleQueryAgentDetails( client_connection_block_t* ccb, ra_command_t* command )
{
	ra_message_t *replyMessage;
	ra_command_t *replyCommand;
	agent_t *agent;
	process_t *process;
	cctl_state_data_t*  cctlData;

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

	/* Locate the process */
	process = findProcess(cctlData, command->info.query_agent_details.processId);

	if(process==NULL) {
		/* Should return an error here */
		TPTP_LOG_DEBUG_MSG1(ccb->stateData, "Unable to find process (%u) for query agent details", command->info.query_agent_details.processId);
		return FALSE;
	}


	/* Locate the active agent */
	agent=ra_agentFindByProcess(process,
					&command->info.query_agent_details.agent,
					TRUE);

	/* Did the active agent exist? */
	if(agent==NULL) {
		/* Should return an error here */
		TPTP_LOG_DEBUG_MSG1(ccb->stateData, "Unable to find agent (%s) for query agent details", command->info.query_agent_details.agent.data);
		return FALSE;
	}

	replyMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
	replyCommand=ra_addCommandToMessage(replyMessage, NULL);

	/* Fill the details into the message */
	replyCommand->tag=RA_AGENT_DETAILS;
	replyCommand->info.agent_details.context=command->info.query_agent_details.context;
	replyCommand->info.agent_details.processId=command->info.query_agent_details.processId;
	ra_copyRASTRING(&replyCommand->info.agent_details.processUUID, &process->uuid);
	ra_copyRASTRING(&replyCommand->info.agent_details.agent, &agent->agentName);
	ra_copyRASTRING(&replyCommand->info.agent_details.agentUUID, &agent->agentUUID);
	ra_copyRASTRING(&replyCommand->info.agent_details.agentType, &agent->agentType);

	/* Forward the message */
	ra_forwardMessage(replyMessage, ccb);
	ra_destroyMessage(replyMessage, TRUE);

	return TRUE;
}

BOOL handleRegisterAgentNotification( client_connection_block_t* ccb, ra_command_t* command )
{
	agent_t *agent;
	process_t *process;
	cctl_state_data_t*  cctlData;

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

	process = findProcess(cctlData, command->info.register_agent_notification.processId);

	if(process) {

		/* Initialize agent */
		agent=ra_agentCreate(process, FALSE);
		ra_copyRASTRING(&agent->agentName, &command->info.register_agent_notification.agent);
		agent->attached=TRUE;
		agent->client=(client_t*)tptp_malloc(sizeof(client_t));
		if (agent->client != NULL ) {
			agent->client->clientID = ccb->clientID;
		}
		agent->context=command->info.launch_process.context;
		//ra_logServiceMessage(__FILE__, __LINE__, RA_INFORMATION, "Adding agent interest: %s",agent->agentName.data);
	}
	return TRUE;
}

BOOL handleAttachToAgent( client_connection_block_t* ccb, ra_command_t* command )
{
	agent_t*            agent;
	process_t*          process;
	cctl_state_data_t*  cctlData;

	cctlData = (cctl_state_data_t*)ccb->stateData->implData;
	if ( cctlData == NULL )
	{
		sendErrorMessage("RAC003",
						 "Memory allocation error",
						 command->info.attach.context,
						 ccb);
		return FALSE;
	}


	/* Find the process */
	process = findProcess(cctlData, command->info.attach.processId);

	/* If no such process return an error */
	if ( !process ) { /* 215066 */
		/* TO_DO:  Set proper error number */
		sendErrorMessage("RAC007",
						 "No such agent",
						 command->info.attach.context,
						 ccb);
		return FALSE;
	}

	/* Look in the active agent list */
	agent=ra_agentFindByProcess(process,
					&command->info.attach.agent,
					TRUE);


	/* Inactive agent */
	if(agent==NULL) {
		/* Look in the inactive table */
		agent=ra_agentFindByProcess(process,
						&command->info.attach.agent,
						FALSE);

		/* No such agent, create inactive agent with client association */
		if(agent==NULL) {
			agent = ra_agentCreate(process, FALSE);
			ra_copyRASTRING(&agent->agentName, &command->info.attach.agent);
		}
		if(ccb) {/* 215066 */
			ra_connectAgentToClient(agent, ccb);
		}

		agent->context=command->info.attach.context;
	}
	else if(!agent->attached) {
		tptp_int32 ret;

		if(ccb) {/* 215066 */
			ra_connectAgentToClient(agent, ccb);

			/* Send a command to the AC to attach based on the 'agent' structure */
			ret = getAgentByToken( ccb->stateData, ccb, agent );
			if ( ret == 0 )
			{
				ra_message_t *replyMessage;
				ra_command_t *replyCommand;

				agent->context=command->info.attach.context;

				replyMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
				replyCommand=ra_addCommandToMessage(replyMessage, NULL);

				TPTP_LOG_INFO_MSG(ccb->stateData, "Preparing Agent active");

				replyCommand->tag=RA_AGENT_ACTIVE;
				replyCommand->info.agent_active.context=agent->context;
				replyCommand->info.agent_active.processId=process->pid;
				ra_copyRASTRING(&replyCommand->info.agent_active.processUUID, &process->uuid);
				ra_copyRASTRING(&replyCommand->info.agent_active.agent, &agent->agentName);
				ra_copyRASTRING(&replyCommand->info.agent_active.agentUUID, &agent->agentUUID);
				ra_copyRASTRING(&replyCommand->info.agent_active.agentType, &agent->agentType);

				// Forward the message
	    		ra_forwardMessage( replyMessage, ccb );

				ra_destroyMessage(replyMessage, TRUE);
			}
		}
	}
	else {
		/* Agent exists and is currently attached, inform client */
		if(ccb) { /* 215066 */
			sendErrorMessage("RAC008",
						 "Agent already attached to another client",
						 command->info.attach.context,
						 ccb);
		} /* 215066 */
		return FALSE;
	}

	return TRUE;
}

BOOL handleDetachFromAgent( client_connection_block_t* ccb, ra_command_t* command )
{

	char          releaseAgentFormat[] = "<Cmd src=\"%d\" dest=\"%d\" ctxt=\"%d\"><releaseAgent iid=\"org.eclipse.tptp.agentManager\"><agentID>%lu</agentID></releaseAgent></Cmd>";
	char          releaseAgentCommand[8192];
	unsigned int  contextID = baseTL_getNextContextID();
	cctl_state_data_t*  cctlData;
	process_t *process;
	agent_t *agent;
	tptp_int32 status;

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

	status =  findProcessAndAgent(command->info.detach.processId, 
			&command->info.detach.agent, TRUE, cctlData->processList, &agent, &process);

	if (!process) {
		return FALSE;
	}
	/* Check the inactive agents */
	if(agent==NULL) {
		agent=ra_agentFindByProcess(process, &command->info.detach.agent, FALSE);
		if(agent==NULL) {
			return FALSE;
		}
	}

	/* Is this client attached to the agent */

	if(agent->client->clientID !=ccb->clientID) {
		return FALSE;
	}

	//Release agent will all handle the remove of the data path.
	sprintf( releaseAgentCommand, releaseAgentFormat, ccb->clientID, AGENT_MANAGER, contextID, agent->agentID);

	/* there is no response to this command being sent... other than an error... which still is not passed  */
	/* back to the client  											*/
	forwardXmlCommand( ccb->stateData, releaseAgentCommand);

	ra_disconnectAgentFromClient(agent, ccb);

	agent->IPCBufCreated = FALSE;
	agent->logging = FALSE;

	return TRUE;
}

BOOL handleStartMonitoring( client_connection_block_t* ccb, ra_command_t* command )
{
	agent_t *agent;
	process_t *process;
	char *buffname=NULL;        /* shared memory buffer name */
	cctl_state_data_t*  cctlData;
	tptp_int32 status = -1;
	tptp_int32 rc = -1;
	tptp_int32 flags;
	tptp_int32 dataConnectionID;
	struct sockaddr_storage * addr;
	client_data_connection_block_t *dcb;
	SOCKET dataSock;

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

	status =  findProcessAndAgent(command->info.start_monitor_remote.processId, 
			&command->info.start_monitor_remote.agent, TRUE, cctlData->processList, &agent, &process);

	if(process==NULL) {
		return FALSE;
	}

	/* Agent unavailable, return error string */
	if(agent==NULL) { /* 215066 */
		sendErrorMessage("RAC001",
						 "No such agent",
					 	 command->info.start_monitor_remote.context,
						 ccb);
		return FALSE;
	}

	/* Set the flags according to the configured data transfer style */
	/* TODO: Put code here to set up logging data to a file */
	if ( cctlData->isDataMultiplexed )
	{
		flags = TPTP_DATA_PATH_RECEIVE | TPTP_DATA_PATH_MULTIPLEXED;

		if (agent->IPCBufCreated == FALSE) {
			dcb = (client_data_connection_block_t*)tptp_malloc( sizeof(client_data_connection_block_t) );
			if ( dcb == NULL ) {
				sendErrorMessage("RAC003",
								 "Memory allocation error",
								 command->info.start_monitor_remote.context,
								 ccb);
				return FALSE;
			}
			dcb->sock    = (SOCKET)NULL;
			ra_copyRASTRING( &dcb->agentName, &agent->agentName );
			dcb->context = agent->context;
			dcb->pid     = agent->process->pid;
			dcb->ccb     = ccb;

			rc = ccb->stateData->ac.addDataConnection( ccb->stateData->ac.cmo, ccb->stateData->transportID, flags, &dataConnectionID );
			baseTL_addDataConnectionEntry( ccb->stateData, dataConnectionID, (void*)dcb );

			agent->dataConnectionID = dataConnectionID;
			agent->IPCBufCreated = TRUE;
			agent->logging = TRUE;
		}
	}
	else
	{
		flags = TPTP_DATA_PATH_RECEIVE | TPTP_DATA_PATH_DIRECT_SOCKET;

		if (agent->IPCBufCreated == FALSE) {

			getClientIPAddr(ccb, &addr);
			dataSock = openClientDataSocketAddr(addr, command->info.start_monitor_remote.port);
			
			if (dataSock == -1) {
				sendErrorMessage("RAC009",
				 "Start Monitoring failed due to data channel socket connection failure.", 
				 command->info.start_monitor_remote.context,
				 ccb);
				return FALSE;
			}

			dcb = (client_data_connection_block_t*)tptp_malloc( sizeof(client_data_connection_block_t) );
			if ( dcb == NULL ) {
				sendErrorMessage("RAC003",
								 "Memory allocation error",
								 command->info.start_monitor_remote.context,
								 ccb);
				return FALSE;
			}

			dcb->sock    = dataSock;

			ra_copyRASTRING( &dcb->agentName, &agent->agentName );
			dcb->context = agent->context;
			dcb->pid     = agent->process->pid;
			dcb->ccb     = ccb;

			rc = ccb->stateData->ac.addDirectDataConnection( ccb->stateData->ac.cmo, ccb->stateData->transportID, flags, dataSock, &dataConnectionID );
			baseTL_addDataConnectionEntry( ccb->stateData, dataConnectionID, (void*)dcb );

			agent->dataConnectionID = dataConnectionID;
			agent->IPCBufCreated = TRUE;
			agent->logging = TRUE;
		}

	}

	status = establishDataPath(ccb->stateData, ccb, agent, agent->dataConnectionID, flags);

	return TRUE;
}

BOOL handleStopMonitoring( client_connection_block_t* ccb, ra_command_t* command )
{
	agent_t *agent;
	process_t *process;
	cctl_state_data_t*  cctlData;
	tptp_int32 status = -1;

	PID pid = command->info.stop_monitor.processId;

	cctlData = (cctl_state_data_t*)ccb->stateData->implData;
	if ( cctlData == NULL )
	{
		sendErrorMessage("RAC003",
						 "Memory allocation error",
						 command->info.stop_monitor.context,
						 ccb);
		return FALSE;
	}

	status =  findProcessAndAgent(command->info.stop_monitor.processId, 
			&command->info.stop_monitor.agent, TRUE, cctlData->processList, &agent, &process);

	if(agent==NULL) {
		return FALSE;
	}

	status = releaseDataPath(ccb->stateData, ccb, agent);

	agent->logging=FALSE;
	return TRUE;
}

BOOL handleManageFile( client_connection_block_t* ccb, ra_command_t* command )
{
	switch (command->info.manage_file.operation)
	{
	case RA_GET_FILE:
	case RA_PUT_FILE:
		TPTP_LOG_INFO_MSG1( ccb->stateData, "Received %s command from client", (command->info.manage_file.operation == RA_GET_FILE) ? "GET_FILE" : "PUT_FILE"); 
		sendFileServerPort(command, ccb);
		break;
	case RA_DELETE_FILE:
		TPTP_LOG_INFO_MSG( ccb->stateData, "Received DELETE_FILE command from client"); 
		unlink(command->info.manage_file.filename.data);
		break;
	}
	return TRUE;
}

/** Method to send the configured file port to the client on request */

static int sendFileServerPort (ra_command_t *command, client_connection_block_t* ccb)
{
	ra_message_t *replyMessage;
	ra_command_t *replyCommand;
	char buffer[33];
	static int jobKey = 0;

	/* Formulate the reply to the client */
	replyMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
	replyCommand=ra_addCommandToMessage(replyMessage, NULL);

	/* We send a resource location message back to the client */
	replyCommand->info.resource_location.context = command->info.manage_file.context;
	replyCommand->tag = RA_RESOURCE_LOCATION;

	/* Giridhar S - 23/04/2005; Secure File Transfer enablement */
	replyCommand->info.resource_location.port = ((cctl_state_data_t*)ccb->stateData->implData)->filePort;

	/* TODO: The job key currently contains a dummy value. Is it really needed now? */
	sprintf(buffer, "%d", jobKey++);
	ra_createRASTRING(&replyCommand->info.resource_location.jobKey, buffer);

	/* Send a message back to the client with configured file server port */
	ra_forwardMessage(replyMessage, ccb);

	/* Free used memory */
	ra_destroyMessage(replyMessage, TRUE);
	return 0;
}

BOOL handleGetPropertyList( client_connection_block_t* ccb, ra_command_t* command )
{
	char        cmdFormat[] = "<Cmd src=\"%d\" dest=\"%d\" ctxt=\"%d\"><getPropertyList iid=\"org.eclipse.tptp.agentManager\"><name>%s</name><type>%s</type></getPropertyList></Cmd>";
	char*       cmd;
	int         cmdSize;
	tptp_uint32 context;
	get_property_list_context_t* getPropertyListContextData;

	/* Calculate the size of the command we're building */
	cmdSize = strlen(cmdFormat) + 24 
	            + command->info.query_property_list.name.length 
	            + command->info.query_property_list.type.length;

	/* Allocate memory for the command */
	cmd	= tptp_malloc( cmdSize );
	if ( cmd == NULL )
	{
		sendErrorMessage("RAC003",
						 "Memory allocation error",
						 command->info.query_property_list.context,
						 ccb);
		return FALSE;
	}

	/* Format the command */
	context = baseTL_getNextContextID();
	sprintf( cmd, cmdFormat, ccb->clientID, AGENT_MANAGER, context, 
		                     command->info.query_property_list.name.data,
	                         command->info.query_property_list.type.data );

	/* Save our context relative to the client */
	getPropertyListContextData = (get_property_list_context_t*)tptp_malloc(sizeof(get_property_list_context_t));
	if ( getPropertyListContextData == NULL )
	{
		tptp_free(cmd);
		sendErrorMessage("RAC003",
						 "Memory allocation error",
						 command->info.query_property_list.context,
						 ccb);
		return FALSE;
	}
	getPropertyListContextData->contextDataType = GET_PROPLIST_CONTEXT_TYPE;
	getPropertyListContextData->clientContext = command->info.query_property_list.context;
	baseTL_storeContextData( ccb->stateData, context, (void*)getPropertyListContextData );

	/* Send the command, when we get a response, we will send the response to the client */
	forwardXmlCommand( ccb->stateData, cmd );

	return TRUE;
}

BOOL handlePeerRequestMonitor( client_connection_block_t* ccb, ra_command_t* command )
{
	agent_t *targetAgent;
	unsigned char address[4];
	BOOL isLocalMachine=FALSE;
	int port = RA_CTL_PORT_NUM_SERVER; /* Bug 77768 */
	tl_control_connection_t* targetConnectionEntry;
	client_connection_block_t* targetCcb;

	/* Extract the IP address from the command. THIS IS IPV4 SPECIFIC */
	address[0]=command->info.controller_request_monitor.peerNode.data[0];
	address[1]=command->info.controller_request_monitor.peerNode.data[1];
	address[2]=command->info.controller_request_monitor.peerNode.data[2];
	address[3]=command->info.controller_request_monitor.peerNode.data[3];

	if(command->tag == RA_CONTROLLER_REQUEST_MONITOR_PORT) {
		port = command->info.agent_request_monitor_port.port; /* bug 77768: Server port used on the second RAC */
	}

	/* Find the peer agent */
	/* Locate the active agent.  If a PID was provided search by PID, else the peerAgent is a UUID */
	if(command->info.agent_request_monitor.peerProcessId) {
		process_t* process;
		cctl_state_data_t* cctlData = (cctl_state_data_t*)ccb->stateData->implData;

		if ( !findProcessAndAgent( command->info.agent_request_monitor.peerProcessId,
					&command->info.agent_request_monitor.peerAgent,
					TRUE,
					cctlData->processList,
					&targetAgent,
					&process ) )
		{
			targetAgent = NULL;
		}
	}
	else {
		targetAgent=ra_agentFindByUUID(ra_getActiveProcessList(), &command->info.agent_request_monitor.peerAgent, TRUE);
	}

	/* Did the active agent exist? We need to send an error back to the
	   requesting RAC.
	*/
	if(targetAgent==NULL) {
		ra_message_t *forwardMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
		command->tag=RA_PEER_UNREACHABLE;
		TPTP_LOG_WARNING_MSG2( ccb->stateData, "Cannot connect to local agent %s on pid %d", command->info.agent_request_monitor.peerAgent.data, command->info.agent_request_monitor.peerProcessId);
		ra_addCommandToMessage(forwardMessage, command);
		ra_forwardMessage(forwardMessage, ccb);
		ra_destroyMessage(forwardMessage, FALSE);
		return FALSE;
	}

	/* Change the tag on the command and return the result to the remote Client */
	if(targetAgent->attached && targetAgent->client) {
		ra_message_t *forwardMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);

		/* Bug 77768: check whether the port is default or not */
		if(port == RA_CTL_PORT_NUM_SERVER) {
			command->tag=RA_CONTROLLER_REQUEST_MONITOR;

			/* Use the context of the target peer */
			command->info.controller_request_monitor.context=targetAgent->context;

			/* Fill in the local agents information */
			command->info.controller_request_monitor.peerProcessId=targetAgent->process->pid;
			ra_copyRASTRING(&command->info.controller_request_monitor.peerAgent, &targetAgent->agentName);

			/* Put In our local address. RKD:  This is incorrect  */
			/*    AWK:  I'm leaving it anyway because we want to match RAC behavior */
			ra_createRASTRING3(&command->info.controller_request_monitor.peerNode, (const char*)address, 4);

			/* Fill in the agents information */
			command->info.controller_request_monitor.peerProcessId=targetAgent->process->pid;
			ra_copyRASTRING(&command->info.controller_request_monitor.peerAgent, &targetAgent->agentName);
		}
		else { /* Otherwise use a new command which supports port */
			command->tag=RA_CONTROLLER_REQUEST_MONITOR_PORT;

			/* Use the context of the target peer */
			command->info.controller_request_monitor_port.context=targetAgent->context;

			/* Fill in the local agents information */
			command->info.controller_request_monitor_port.peerProcessId=targetAgent->process->pid;
			ra_copyRASTRING(&command->info.controller_request_monitor_port.peerAgent, &targetAgent->agentName);

			/* Put In our local address. RKD:  This is incorrect */
			/*    AWK:  I'm leaving it anyway because we want to match RAC behavior */
			ra_createRASTRING3(&command->info.controller_request_monitor_port.peerNode, (const char*)address, 4);

			/* Fill in the agents information */
			command->info.controller_request_monitor_port.peerProcessId=targetAgent->process->pid;
			ra_copyRASTRING(&command->info.controller_request_monitor_port.peerAgent, &targetAgent->agentName);

			/* Fill in the server port on the remote RAC */
			command->info.controller_request_monitor_port.port = port; /* Bug 77768 */
		}

		ra_addCommandToMessage(forwardMessage, command);
		TPTP_LOG_DEBUG_MSG( ccb->stateData, "Forwarding remote peer monitor request to client");

		targetConnectionEntry = (tl_control_connection_t*) tableGet(ccb->stateData->controlConnectionTable, targetAgent->client->clientID) ;
		if ( targetConnectionEntry == NULL )
		{
			/* Unrecognized connection */
			ra_message_t *forwardMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
			command->tag=RA_PEER_UNREACHABLE;
			TPTP_LOG_WARNING_MSG2( ccb->stateData, "Unable to find the client associated with local agent %s on pid %d", command->info.agent_request_monitor.peerAgent.data, command->info.agent_request_monitor.peerProcessId);
			ra_addCommandToMessage(forwardMessage, command);
			ra_forwardMessage(forwardMessage, ccb);
			ra_destroyMessage(forwardMessage, FALSE);
			return FALSE;
		}
		targetCcb = (client_connection_block_t*)targetConnectionEntry->implData;
		if ( targetCcb == NULL )
		{
			/* Missing client connection block */
			ra_message_t *forwardMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
			command->tag=RA_PEER_UNREACHABLE;
			TPTP_LOG_WARNING_MSG2( ccb->stateData, "Unable to resolve client connection information for local agent %s on pid %d is not attached and monitored", command->info.agent_request_monitor.peerAgent.data, command->info.agent_request_monitor.peerProcessId);
			ra_addCommandToMessage(forwardMessage, command);
			ra_forwardMessage(forwardMessage, ccb);
			ra_destroyMessage(forwardMessage, FALSE);
			return FALSE;
		}

		ra_forwardMessage(forwardMessage, targetCcb);
		ra_destroyMessage(forwardMessage, FALSE);
	}
	else {
		/* We need to tell the requesting Agent that the peer is not monitored */
		ra_message_t *forwardMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
		command->tag=RA_PEER_UNREACHABLE;
		TPTP_LOG_WARNING_MSG2( ccb->stateData, "Local agent %s on pid %d is not attached and monitored", command->info.agent_request_monitor.peerAgent.data, command->info.agent_request_monitor.peerProcessId);
		ra_addCommandToMessage(forwardMessage, command);
		ra_forwardMessage(forwardMessage, ccb);
		ra_destroyMessage(forwardMessage, FALSE);
		return FALSE;
	}

	return TRUE;
}

BOOL handlePeerUnreachable( client_connection_block_t* ccb, ra_command_t* command )
{
	/* TODO: figure out where this came from and propogate it backward */
	return TRUE;
}

BOOL forwardCommandToAgent( client_connection_block_t* ccb, ra_command_t* command )
{
	cctl_state_data_t*  cctlData;
	agent_t*            agent;
	process_t*          process;
	tptp_int32          status;
	char*               buffer;

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

	status =  findProcessAndAgent(command->info.custom_command.processId, 
			&command->info.custom_command.agent, TRUE, cctlData->processList, &agent, &process);

	if ((process==NULL) || (agent==NULL)){
		return FALSE;
	}

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

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

	tptp_free( buffer );

	return TRUE;
}


