/*******************************************************************************
 * Copyright (c) 2005, 2009 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
 *    Kevin O'Leary, Intel - Extended Implementation
 *    Spundun Bhatt (spundun@gmail.com), with the support and encouragement of the University of Southern California Information Sciences Institute Distributed Scalable Systems Division.
 *
 * $Id: Connect2AC.c,v 1.65 2009/09/24 21:02:04 jwest Exp $
 *
 *******************************************************************************/ 

#include "tptp/TPTPTypes.h"
#include "tptp/ACFlags.h"
#include "tptp/TPTPUtils.h"
#include "tptp/TPTPSupportTypes.h"
#include "tptp/TransportSupport.h"
#include "tptp/TPTPBaseTL.h"
#include "tptp/BaseTLLog.h"
#include "AgentCTL.h"
#include "Connect2AC.h"
#include "AgentCmdHandlers.h"
#include "RACAgentSupport.h"
#include "tptp/dime.h"

#ifdef __linux__
	#include <sys/stat.h>
#endif

#include <fcntl.h>
#include <stdio.h>

#ifdef _WIN32
#include <io.h>
#endif

#define CMD_BUFFER_LENGTH 8192 /* TODO: Use a universal definition */
#define MAX_AGENT_NAME    1024 /* TODO: Use a universal definition */

#define RA_SHM_BUF_SIZE_BC_DEFAULT   (131072*8) /* shared memory buffer size = 1 Mb */

tptp_int32 processControllerDestID = -1;

tptp_int32 forwardXmlCommand( tl_state_data_t* stateData, tptp_string* command )
{
	/* Check to make sure we have a good function pointer to send the message */
	if ( stateData->controlPartner.processMessage == NULL )
	{
		return -1;
	}

	/* Send the command */
	stateData->controlPartner.processMessage( stateData->controlPartner.nexto, strlen(command), command );

	return 0;
}

/*******************************************************************************/ 
/*******************************************************************************/ 

/* Create a connection with the Agent Controller corresponding to this agent */
tptp_int32 registerAgent( tl_state_data_t* stateData, agent_t* agent )
{
	tptp_int32  rc;
	tptp_uint32 connectionID;
	char        commandFormat[] = "<Cmd src=\"%ld\" dest=\"%ld\" ctxt=\"%ld\"><registerAgent iid=\"org.eclipse.tptp.agentManager\"><agentID>%ld</agentID><processID>%lu</processID><agentName>%s</agentName></registerAgent></Cmd>";
	char        command[CMD_BUFFER_LENGTH];
	char        agentNameFormat[] = "org.eclipse.tptp.legacy.%s";
	char        fullAgentName[MAX_AGENT_NAME];

	/* 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 agent: %s", agent->agentName.data );
		return rc;
	}

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

	/* Save this ID for later use */
	agent->agentID = connectionID;

	/* Create an agent name with our standard legacy qualifier */
	sprintf( fullAgentName, agentNameFormat, agent->agentName.data );

	/* Build a command to register this agent with the AC */
	sprintf( command, commandFormat, connectionID, AGENT_MANAGER, baseTL_getNextContextID(), connectionID, agent->process->pid, fullAgentName );

	rc = forwardXmlCommand( stateData, command );
	if ( rc != 0 )
	{
		TPTP_LOG_ERROR_MSG1( stateData, "No processMessage function available to send agent registration: %s", agent->agentName.data );
	}

	return 0;
}

/*******************************************************************************/ 
/*******************************************************************************/ 

tptp_int32 deregisterAgent( tl_state_data_t* stateData, agent_t* agent )
{
	tptp_int32  rc;
 	char commandFormat[] = "<Cmd src=\"%ld\" dest=\"%ld\" ctxt=\"%ld\"><deregisterAgent iid=\"org.eclipse.tptp.agentManager\"><agentID>%ld</agentID></deregisterAgent></Cmd>";
	char command[CMD_BUFFER_LENGTH];

	/* Build a command to deregister this agent with the AC */
	sprintf( command, commandFormat, agent->agentID, AGENT_MANAGER, baseTL_getNextContextID(), agent->agentID );

	rc = forwardXmlCommand( stateData, command );
	if ( rc != 0 )
	{
		TPTP_LOG_ERROR_MSG1( stateData, "No processMessage function available to send agent deregistration: %s", agent->agentName.data );
		return rc;
	}

	return 0;
}

tptp_int32 getAgent(  tl_state_data_t* stateData, agent_t* agent, char *agentName, tptp_int32* agentID )
{
    tptp_int32           status;
	char                 getAgentFormat[] = "<Cmd src=\"%d\" dest=\"%d\" ctxt=\"%d\"><getAgent iid=\"org.eclipse.tptp.agentManager\"><agentName>%s</agentName><flags>%lu</flags></getAgent></Cmd>";
	char                 getAgentCommand[8192];
	unsigned int         contextID = baseTL_getNextContextID();
	get_agent_context_t* getAgentContextData;

	/* Initialize the context data for this command */
	getAgentContextData = (get_agent_context_t*)tptp_malloc(sizeof(get_agent_context_t));
	if ( getAgentContextData == NULL )
	{
		TPTP_LOG_ERROR_MSG( stateData, "Out of memory while getting agent" );
		return TPTP_ERROR_OUT_OF_MEMORY;
	}

	getAgentContextData->contextDataType = GET_AGENT_CONTEXT_TYPE;
	tptp_initializeSemaphore( &getAgentContextData->completionSemaphore );
	getAgentContextData->agentID = -1;
	getAgentContextData->status = ACTL_STATUS_PENDING;
	baseTL_storeContextData( stateData, contextID, (void*)getAgentContextData );

	sprintf( getAgentCommand, getAgentFormat, agent->agentID, AGENT_MANAGER, contextID, agentName, TPTP_CONTROLLER_ACCESS);

	/* Send the command */
	forwardXmlCommand( stateData, getAgentCommand );

	/* Wait for the response */
	tptp_waitSemaphore( &getAgentContextData->completionSemaphore );
	if ( getAgentContextData->status == ACTL_STATUS_PENDING )
	{
		/* This probably means our wait failed */
		/* TODO: Log an error? */
		getAgentContextData->status = -1;
	}
	else if ( getAgentContextData->status == 0 )
	{
		/* Success! */
		*agentID = getAgentContextData->agentID;
	}

	status = getAgentContextData->status;

	/* Clean up */
	tptp_deleteSemaphore( &getAgentContextData->completionSemaphore );
	baseTL_releaseContextData( stateData, contextID );

	return status;
}

BOOL getProcessUUID( tl_state_data_t* stateData, agent_t* agent, PID pid, tptp_string** uuid)
{
	char          getProcessFormat[] = "<Cmd src=\"%d\" dest=\"%d\" ctxt=\"%d\"><getProcessUUID iid=\"org.eclipse.tptp.processController\"><processID>%lu</processID></getProcessUUID></Cmd>";
	char          getProcessCommand[8192];
	unsigned int  contextID = baseTL_getNextContextID();
	tptp_int32    status;
	get_process_uuid_context_t* getProcessUUIDContextData;

	getProcessUUIDContextData = (get_process_uuid_context_t*)tptp_malloc(sizeof(get_process_uuid_context_t));
	if ( getProcessUUIDContextData == NULL )
	{
		TPTP_LOG_ERROR_MSG( stateData, "Out of memory while getting process UUID" );
		return TPTP_ERROR_OUT_OF_MEMORY;
	}

	getProcessUUIDContextData->contextDataType = GET_PROCESS_UUID_CONTEXT_TYPE;
	tptp_initializeSemaphore( &getProcessUUIDContextData->completionSemaphore );
	getProcessUUIDContextData->status = ACTL_STATUS_PENDING;
	baseTL_storeContextData( stateData, contextID, (void*)getProcessUUIDContextData );

	if (processControllerDestID == -1) {
		status = getAgent(stateData, agent, "ProcessController", &processControllerDestID);
	}

	sprintf( getProcessCommand, getProcessFormat, agent->agentID, processControllerDestID, contextID, pid);

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

	tptp_waitSemaphore( &getProcessUUIDContextData->completionSemaphore );
	if ( getProcessUUIDContextData->status == ACTL_STATUS_PENDING )
	{
		/* This probably means our wait failed */
		/* TODO: Log an error? */
		getProcessUUIDContextData->status = -1;
	}
	else if ( getProcessUUIDContextData->status == 0 )
	{
		/* Success! */
		*uuid = tptp_malloc(strlen(getProcessUUIDContextData->uuid) +1);
		strcpy(*uuid, getProcessUUIDContextData->uuid);
		tptp_free(getProcessUUIDContextData->uuid);
	}

	status = getProcessUUIDContextData->status;

	/* Clean up */
	tptp_deleteSemaphore( &getProcessUUIDContextData->completionSemaphore );
	baseTL_releaseContextData( stateData, contextID );

	return status;
}

/*******************************************************************************/ 
/*******************************************************************************/ 

tptp_int32 handleOutgoingMessage( tl_state_data_t* stateData, agent_t* agent, tptp_string* pCmdBlock )
{
	tptp_int32     ret;
	tptp_int32     sourceID;
	tptp_int32     context;
	char*          interfaceID = NULL;
	char*          cmdName = NULL;
	tptp_list_t*   paramList;

	/* Parse the XML into useful blocks */
	ret = parseCommand( pCmdBlock, &sourceID, &context, &interfaceID, &cmdName, &paramList );
	if ( ret != 0 )
	{
		/* We can't respond (because we don't know where to send it)   */
		/*    so log it as an error                                    */
		TPTP_LOG_ERROR_MSG1( stateData, "Error parsing incoming command: %s", pCmdBlock );
		return ret;
	}
	else
	{

		ret = processCommand( stateData, agent, pCmdBlock, sourceID, context, interfaceID, cmdName, paramList );

		/* Free the resources allocated by parseCommand */
		if (interfaceID != NULL) tptp_free( interfaceID );
		if (cmdName != NULL) tptp_free( cmdName );
		tptp_list_clear( paramList );
		tptp_free( paramList );
		return ret;
	}

	return 0;
}

/*******************************************************************************/ 

tptp_int32 handlePropertyList( tl_state_data_t*           stateData, 
				               agent_t*                   agent, 
                               tptp_int32                 contextID,
                               tptp_list_t*               paramList )
{
	tptp_int32                    ret;
	actl_get_proplist_context_t*  gplContextData;
	tptp_list_t                   propertyList;
	ra_message_t*                 replyMessage;
	ra_command_t*                 replyCommand;

	/* Get the context data for this command */
	ret = baseTL_getContextData(stateData, contextID, (void**)&gplContextData);
	if ( (ret != 0) || (gplContextData == NULL) )
	{
		TPTP_LOG_ERROR_MSG( stateData, "Missing context data in propertyList response" );
		return -1;
	}

	if ( gplContextData->contextDataType != GET_PROPLIST_CONTEXT_TYPE )
	{
		/* Normally, we free the context data we find, but since this didn't match
		   our expectation, we can't be sure we should free it.  Leaks are better than crashes */
		TPTP_LOG_ERROR_MSG( stateData, "Unexpected context data in propertyList response" );
		return -1;
	}

	/* Get the properties */
	ret = getStringListParam( "property", paramList, &propertyList );


	replyMessage = ra_createMessage(RA_CONTROL_MESSAGE, 0);
	replyCommand = ra_addCommandToMessage(replyMessage, NULL);
	replyCommand->tag = RA_PROPERTY_LIST;
	replyCommand->info.property_list.context = gplContextData->agentContext;

	if ( (ret != 0) || (propertyList.count == 0) )
	{
		replyCommand->info.property_list.entries.length = 0;
	}
	else
	{
		tptp_node_t* node;
		int          i = 0;

		for ( node = propertyList.head; node != NULL; node = node->next )
		{
			char* name;
			char* type;
			char* value;

			ret = parsePropertyListEntry( (char*)node->data, &name, &type, &value );
			if ( ret != 0 )
			{
				((ra_agentConfigEntry_t**)replyCommand->info.property_list.entries.data)[i] = (ra_agentConfigEntry_t*)tptp_malloc(sizeof(ra_agentConfigEntry_t));
				BZERO(((ra_agentConfigEntry_t**)replyCommand->info.property_list.entries.data)[i], sizeof(ra_agentConfigEntry_t));
				ra_createRASTRING(&((ra_agentConfigEntry_t**)replyCommand->info.property_list.entries.data)[i]->name, name);
				ra_createRASTRING(&((ra_agentConfigEntry_t**)replyCommand->info.property_list.entries.data)[i]->type, type);
				ra_createRASTRING(&((ra_agentConfigEntry_t**)replyCommand->info.property_list.entries.data)[i]->value, value);
				i++;

				tptp_free(name);
				tptp_free(type);
				tptp_free(value);
			}


		}
	}

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

	/* Free our context data */
	tptp_free( gplContextData );

	return 0;
}

/*******************************************************************************/ 

int MultiplexingDataProcessorFunc(void *pData, int dataLen, void *pArgs)
{
	actl_data_connection_block_t* dcb = (actl_data_connection_block_t *) pArgs ;
	tl_data_connection_t*         dataConnection = (tl_data_connection_t *) tableGet(dcb->stateData->dataConnectionTable, dcb->partnerID);
  	char*                         current;
  	char*                         temp;

	unsigned long magicNumber = 0xFEEDC0DE;
	unsigned long terminationCode = 0xDEADC0DE;

	if( !(dataLen>0 && pData) ) {
		return -1;
	}

	temp = (char*)tptp_malloc(sizeof(char) * (dataLen + 4));
	if ( temp == NULL ) {
		return -1;
	}
	current = temp;

	/* Add the magic number to the front of the byte array */
	current[0]=(unsigned char)(magicNumber>>24 & 0x000000ff);
	current[1]=(unsigned char)(magicNumber>>16 & 0x000000ff);
	current[2]=(unsigned char)(magicNumber>>8  & 0x000000ff);
	current[3]=(unsigned char)magicNumber;
	current += 4;

	memcpy(current, (char*)pData, dataLen);

	/* Call the send routine */
	/* The data connection entry was computed to be our client routine */
	dataConnection->dataPartner.sendData(
			dataConnection->dataPartner.nexto,
			dataConnection->dataPartner.connectionID,
		       	dataLen + 4, temp);

	tptp_free( temp );

	return 0;
}

/*******************************************************************************/ 

int ACTLDataProcessorFunc(void *pData, int dataLen, void *pArgs)
{
	int dimeLength = sizeof(DIME_HEADER_T);
	static char* newBlock = NULL;
	static int blockLength = 0;
	DIME_HEADER_PTR_T p;
	int rc;

	actl_data_connection_block_t*  dcb = (actl_data_connection_block_t *) pArgs ;

	tl_data_connection_t* dataConnection = (tl_data_connection_t *) tableGet(dcb->stateData->dataConnectionTable, dcb->partnerID);

	/* The terminate timeout thread using this to see if we're really getting data */
	dcb->agent->dataReceived = 1;

	/* Need to add a DIME header block to send to client */
	p = (DIME_HEADER_PTR_T) tptp_malloc(sizeof(DIME_HEADER_T));
	rc = INIT_DIME_HEADER(p);
	p->data_length = (tptp_uint32) dataLen;

	MAKE_NBO_DIME_HEADER((char *)p);

	if ( newBlock == NULL )
	{
		newBlock = (char*)tptp_malloc( dataLen+dimeLength );
		if ( newBlock == NULL ) 
		{
			/* TODO: Log out of memory error */
			return -1;
		}
		
		blockLength = dataLen+dimeLength;
	}

	if ( blockLength < dataLen+dimeLength )
	{
		newBlock = (char*)tptp_realloc( newBlock, dataLen+dimeLength );
		if ( newBlock == NULL ) 
		{
			/* TODO: Log out of memory error */
			return -1;
		}
		
		blockLength = dataLen+dimeLength;
	}

	memcpy(newBlock, p, dimeLength);
	memcpy(newBlock+dimeLength, pData, dataLen);

	/* Call the send routine */
	/* The data connection entry was computed to be our client routine */
	dataConnection->dataPartner.sendData(
			dataConnection->dataPartner.nexto,
			dataConnection->dataPartner.connectionID,
		       	dataLen + dimeLength, newBlock);
		

	return 0 ;
}


THREAD_USER_FUNC_RET_TYPE doAgentListening(LPVOID args) 
{
	int rc = 0;
	actl_data_connection_block_t  *dcb = (actl_data_connection_block_t *) args ;
	mem_map_block_ptr_t   pMemBlockInfo = dcb->memBlock;
	tl_data_connection_t* dataConnection = (tl_data_connection_t *) tableGet(dcb->stateData->dataConnectionTable, dcb->partnerID);
	Lock_t*               pLock;

	if ( dcb->isDataMultiplexed )
	{
  		char                  term[4];
		unsigned long         terminationCode = 0xDEADC0DE;

		dcb->agent->isProcessingData = TRUE;  

		/* Call the function to loop until the shared memory is released */
		rc =  ipcFlushToFunc(pMemBlockInfo, MultiplexingDataProcessorFunc, (void *) dcb);

		/* Inform the client that we are done flushing */
		if ( dataConnection != NULL )
		{
			term[0]=(unsigned char)(terminationCode>>24 & 0x000000ff);
			term[1]=(unsigned char)(terminationCode>>16 & 0x000000ff);
			term[2]=(unsigned char)(terminationCode>>8  & 0x000000ff);
			term[3]=(unsigned char)terminationCode;

			/* Call the send routine */
			/* The data connection entry was computed to be our client routine */
			dataConnection->dataPartner.sendData(
					dataConnection->dataPartner.nexto,
					dataConnection->dataPartner.connectionID,
		       			4, term);
		}

		tptp_postSemaphore( &dcb->agent->dataSemaphore );
		dcb->agent->isProcessingData = FALSE;  
	}
	else
	{
		rc =  ipcFlushToFunc(pMemBlockInfo, ACTLDataProcessorFunc, (void *) dcb) ;
	}

	/* Make sure no one else tries to access the shared memory while we're deleting it */
	pLock = &dcb->stateData->dataConnectionLock;
	tptp_getWriteLock( pLock );

	/* Destroy the shared memory resource and release our conneciton data */
	ipcMemDestroy( pMemBlockInfo );
	dcb->stateData->ac.removeDataConnection( dcb->stateData->ac.cmo, dataConnection->connectionID );
	baseTL_removeDataConnectionEntry( dcb->stateData, dataConnection->connectionID );

	/* baseTL_removeDataConnectionEntry frees the connectionEntry memory
	      but it doesn't do a deep free, so we need to free the dcb here */
	tptp_free( dcb );

	/* Release our lock now that the shared memory is cleaned up */
	tptp_releaseWriteLock( pLock ); // This lock will still be valid because it's in
	                                //   the TL state data.  We had to save the pointer
	                                //   because we're deleting the structure through
	                                //   which we got our pointer to the lock

	return ( 0 );
}


tptp_int32 createLogfile( tl_state_data_t* stateData, agent_t* agent )
{
	actl_data_connection_block_t *pData;
	tptp_int32 rc = -1;
	tptp_int32 receiveFlags = TPTP_DATA_PATH_RECEIVE | TPTP_DATA_PATH_DIRECT_FILE;
	tptp_int32 sendFlags = TPTP_DATA_PATH_SEND | TPTP_DATA_PATH_DIRECT_FILE;
	tptp_int32 dataConnectionID;
	unsigned int         contextID = baseTL_getNextContextID();

	pData = tptp_malloc(sizeof(actl_data_connection_block_t));
	if ( !pData )
	{
		TPTP_LOG_ERROR_MSG( stateData, "Out of memory while binding data connections" );
		return TPTP_ERROR_OUT_OF_MEMORY;
	}
	#ifdef __linux__
		pData->fd = open(agent->logFile.data, O_RDWR|O_CREAT|O_APPEND, S_IRWXU);
	#else
		pData->fd = open(agent->logFile.data, O_RDWR|O_CREAT|O_APPEND);
	#endif
	rc = stateData->ac.addDirectDataConnection( stateData->ac.cmo, stateData->transportID, receiveFlags, pData->fd, &dataConnectionID );
	baseTL_addDataConnectionEntry( stateData, dataConnectionID, (void*)pData );

	rc = bindDataConnections(stateData, agent, dataConnectionID, sendFlags, -1, contextID);

	return 0;
}

tptp_int32 bindDataConnections( tl_state_data_t* stateData, agent_t* agent, tptp_uint32 partnerID, tptp_uint32 flags, tptp_uint32 clientID, tptp_int32 contextID )
{
	int rc;
	tptp_uint32 connectionID;
	tptp_int32  bindContextID = baseTL_getNextContextID();
	char   bindConnections[] = "<Cmd src=\"%ld\" dest=\"%d\" ctxt=\"%ld\"> <bindDataConnections iid=\"org.eclipse.tptp.agentManager\"><dataConnection1>%ld</dataConnection1><dataConnection2>%ld</dataConnection2></bindDataConnections></Cmd>";

	char command[CMD_BUFFER_LENGTH];
	static int ra_shm_buf_num = 0;
#define RA_SHM_BUF_NAME_ROOT  "rabuffer"
	char buffname[256];
	mem_map_block_t* pMemBlockInfo;
	actl_data_connection_block_t *pData;
	tptp_int32 ret;

 	TID threadId;
	HANDLE threadHandle;
	actl_bind_data_context_t* dataRec = (actl_bind_data_context_t*)tptp_malloc( sizeof(actl_bind_data_context_t) );

	if ( !dataRec )
	{
		TPTP_LOG_ERROR_MSG( stateData, "Out of memory while binding data connections" );
		return TPTP_ERROR_OUT_OF_MEMORY;
	}

	dataRec->contextDataType = BIND_DATA_PATHS_CONTEXT_TYPE;
	dataRec->clientID = clientID;
	dataRec->contextID = contextID;

	baseTL_storeContextData( stateData, bindContextID, (void*)dataRec );

	//Only need to bind once... so don't notify the Agent Controller.
	//But we do need to notify the agent to start monitoring again.
	//This is done by simulating a return from dataConnectionsBound.
	//Note- this call must be done after the call to set up the context.
	if (agent->IPCBufCreated == TRUE) {
		startMonitor(stateData, agent, contextID);
		ret = handleDataConnectionsBoundCommand( stateData, agent, bindContextID );
		return ret;
	}

	/* Create shared memory name */
	sprintf(buffname, "%s%d", RA_SHM_BUF_NAME_ROOT, agent->agentID);

	pMemBlockInfo = (mem_map_block_t*)tptp_malloc( sizeof(mem_map_block_t) );
	if ( !pMemBlockInfo )
	{
		TPTP_LOG_ERROR_MSG( stateData, "Out of memory while binding data connections" );
		return TPTP_ERROR_OUT_OF_MEMORY;
	}

	/* Create shared memory block */
	if (agent->IPCBufSize == 0) 
	{
		ipcMemCreate(buffname, pMemBlockInfo, RA_SHM_BUF_SIZE_BC_DEFAULT);
	} else {
		ipcMemCreate(buffname, pMemBlockInfo, agent->IPCBufSize);
	}

	rc = stateData->ac.addDataConnection( stateData->ac.cmo, stateData->transportID, flags, &connectionID );

	/* Start thread to handle streaming Agent data */
	pData = tptp_malloc(sizeof(actl_data_connection_block_t));
	if ( !pData )
	{
		TPTP_LOG_ERROR_MSG( stateData, "Out of memory while binding data connections" );
		return TPTP_ERROR_OUT_OF_MEMORY;
	}

	pData->stateData = stateData;
	pData->partnerID = connectionID;
	pData->memBlock  = pMemBlockInfo;
	pData->agent     = agent;
	pData->flags	 = flags;

	if ( flags & TPTP_DATA_PATH_MULTIPLEXED )
	{
		pData->isDataMultiplexed = TRUE;
	}
	else
	{
		pData->isDataMultiplexed = FALSE;
	}

	/* store away the data connection id so we can reference when we want to stop flusing */
	agent->dataConnectionID = connectionID;
	agent->IPCBufCreated = TRUE;

	//Tell the client before we start our flusher thread.
	rc = startMonitor(stateData, agent, contextID);

	baseTL_addDataConnectionEntry( stateData, connectionID, (void*)pData );

	if ( !(flags & TPTP_DATA_PATH_DIRECT_SOCKET) && !(flags & TPTP_DATA_PATH_DIRECT_FILE))
	{
		rc = tptpStartThread(doAgentListening, 
				(LPVOID) pData, &threadId, &threadHandle);
		CLOSE_THREAD_HANDLE(threadHandle);
	}

	sprintf( command, bindConnections, agent->agentID, AGENT_MANAGER,
		bindContextID, partnerID, connectionID);
	rc = forwardXmlCommand( stateData, command );

	return rc;
}

tptp_int32 dataPathEstablished( tl_state_data_t* stateData, agent_t* agent, tptp_uint32 connectionID, tptp_int32 contextID )
{
	int   rc;
	char  commandFormat[] = "<Cmd src=\"%ld\" dest=\"%ld\" ctxt=\"%ld\"> <dataPathEstablished iid=\"org.eclipse.tptp.dataProvider\"></dataPathEstablished></Cmd>";
	char  command[CMD_BUFFER_LENGTH];

	actl_bind_data_context_t* dataRec;

	rc = baseTL_getContextData( stateData, contextID, (void**)&dataRec);
	if (rc == 0) 
	{
		sprintf( command, commandFormat, connectionID, dataRec->clientID, dataRec->contextID);
		rc = forwardXmlCommand( stateData, command );
		baseTL_releaseContextData( stateData, contextID );
	} 
	else 
	{
		TPTP_LOG_ERROR_MSG1( stateData, "Unable to locate context data to send dataPathEstablished response. Context ID = %d", contextID );
	}
	
	return rc;
}

tptp_int32 handleAgentReference( tl_state_data_t* stateData, 
				agent_t* agent, 
				tptp_int32 contextID,
				tptp_list_t* paramList )
{
	tptp_int32 ret;
	get_agent_context_t*  getAgentContextData;

	/* Get the context data for this command */
	ret = baseTL_getContextData(stateData, contextID, (void**) &getAgentContextData);
	if ( (ret != 0) || (getAgentContextData == NULL) || 
		 (getAgentContextData->contextDataType != GET_AGENT_CONTEXT_TYPE) )
	{
		/* This is unexpected and bad.  The code is designed to signal a semaphore
		      when this response comes in, but the mismatch just identifies means
			  that we can't find the semaphore.
		   Did you add a new command call that doesn't use get_agent_context_t? */
		TPTP_LOG_ERROR_MSG( stateData, "Unexpected context data in agentReference response" );
		getAgentContextData->status = -1;
		return -1;
	}

	/* Get the agent ID */
	ret = getIntegerParam( "agentID", paramList, &getAgentContextData->agentID );
	getAgentContextData->status = 0;

	/* Signal completion */
	tptp_postSemaphore( &getAgentContextData->completionSemaphore );
	
	return 0;
}

tptp_int32 handleGetProcessUUID( tl_state_data_t* stateData, 
				agent_t* agent, 
				tptp_int32 contextID,
				tptp_list_t* paramList )
{
	tptp_int32 ret;
	get_process_uuid_context_t*  getProcessUUIDContextData;

	/* Get the context data for this command */
	ret = baseTL_getContextData(stateData, contextID, (void**) &getProcessUUIDContextData);
	if ( (ret != 0) || (getProcessUUIDContextData == NULL) || 
		 (getProcessUUIDContextData->contextDataType != GET_PROCESS_UUID_CONTEXT_TYPE) )
	{
		/* This is unexpected and bad.  The code is designed to signal a semaphore
		      when this response comes in, but the mismatch just identifies means
			  that we can't find the semaphore.
		   Did you add a new command call that doesn't use get_agent_context_t? */
		TPTP_LOG_ERROR_MSG( stateData, "Unexpected context data in agentReference response" );
		getProcessUUIDContextData->status = -1;
		return -1;
	}

	/* Get the agent ID */
	ret = getStringParam( "processUUID", paramList, &getProcessUUIDContextData->uuid );
	getProcessUUIDContextData->status = 0;

	/* Signal completion */
	tptp_postSemaphore( &getProcessUUIDContextData->completionSemaphore );
	
	return 0;
}

/*******************************************************************************/ 

THREAD_USER_FUNC_RET_TYPE flushingTimeoutThread(LPVOID args)
{
	agent_t* agent = (agent_t *)args;

	if ( agent == NULL )
	{
		return -1;
	}

	while ( TRUE )
	{
		SLEEP(1000);

		if ( !agent->isProcessingData )
		{
			return 0;
		}
		else if ( !agent->dataReceived )
		{
			tptp_postSemaphore( &agent->dataSemaphore );
			return 0;
		}

		agent->dataReceived = 0;
	}

	return 0;
}

/*******************************************************************************/ 

void startFlushingTimeoutThread(agent_t* agent)
{
	TID            threadID;
	HANDLE         threadHandle;

	tptpStartThread( flushingTimeoutThread, (void*)agent, &threadID, &threadHandle);
	CLOSE_THREAD_HANDLE(threadHandle);
}


/*******************************************************************************/ 
/*******************************************************************************/ 

tptp_int32 processCommand( tl_state_data_t* stateData, 
                           agent_t*         agent, 
                           tptp_string*     cmd, 
                           tptp_int32       sourceID, 
                           tptp_int32       context, 
                           tptp_string*     interfaceID, 
                           tptp_string*     cmdName, 
                           tptp_list_t*     paramList )
{
	tptp_int32 ret;

	// Bug 284438 - The agent->agentName value is being freed by another thread, so we cannot pass this into our log statement.
	// TPTP_LOG_DEBUG_MSG3( stateData, "ACTL processing command (%s) from srcID %d, agent %s",
	// cmdName, sourceID, agent->agentName.data);

	/* Check the interface */
	if ( isEqualString( interfaceID, "org.eclipse.tptp.Agent" ) )
	{
		/* Handle base agent commands */
		if ( isEqualString( cmdName, "terminate" ) )
		{
			ret = handleTerminateCommand( stateData, agent );
		}
		else if ( isEqualString( cmdName, "notifyReference" ) )
		{
			tptp_uint32 clientID;

			if (0 != getIntegerParam( "clientID", paramList, &clientID ) )
			{
				TPTP_LOG_ERROR_MSG( stateData, "The notifyReference command arrived without a client ID.");
		
				// If we don't have a valid client ID, we can't do anything with this
				return -1;
			}

			/* Note: there may also be a flags parameter, but we don't use it for compatibility */

			ret = handleNotifyReferenceCommand( stateData, agent, clientID );
		}
		else if ( isEqualString( cmdName, "notifyDereference" ) )
		{
			tptp_uint32 clientID;

			if (0 != getIntegerParam( "clientID", paramList, &clientID ) )
			{
				TPTP_LOG_ERROR_MSG( stateData, "The notifyDereference command arrived without a client ID.");
		
				// If we don't have a valid client ID, we can't do anything with this
				return -1;
			}

			ret = handleNotifyDereferenceCommand( stateData, agent, clientID );
		}
		else
		{
			/* Log this as an unrecognized command */
			TPTP_LOG_ERROR_MSG2( stateData, "ACTL Received unrecognized command: iid=%s, cmd=%s", interfaceID, cmdName );
			return -1;
		}
	}
	else if ( isEqualString( interfaceID, "org.eclipse.tptp.dataProvider" ) )
	{
		/* Handle data provider commands */
		if ( isEqualString( cmdName, "establishDataPath" ) )
		{
			int partnerID;
			int flags;

			if (0 != getIntegerParam( "dataConnectionID", paramList, &partnerID ) )
			{
				TPTP_LOG_ERROR_MSG( stateData, "The establishDataPath command arrived without a dataConnectionID.");
				return -1;
			}

			if (0 != getIntegerParam( "flags", paramList, &flags ) )
			{
				TPTP_LOG_WARNING_MSG( stateData, "The establishDataPath command arrived without a flags parameter.");
				/* We only support one flag setting so if there are no flags we can handle that */
			}
			else if ( (flags != TPTP_DATA_PATH_RECEIVE ) && 
			          (flags != (TPTP_DATA_PATH_MULTIPLEXED | TPTP_DATA_PATH_RECEIVE)) &&
			          (flags != (TPTP_DATA_PATH_DIRECT_SOCKET | TPTP_DATA_PATH_RECEIVE)) )
			{
				TPTP_LOG_WARNING_MSG( stateData, "Unexpected data direction flag sent to legacy agent.");
				/* The client may be able to work with the unidirectional connection we can establish */
			}

			/* We switch the flag from RECEIVE to SEND because of the semantics of this command */
			/* The caller wants to receive, so we must send */
			if ( flags & TPTP_DATA_PATH_DIRECT_SOCKET )
			{
				ret = handleEstablishDataPathCommand( stateData, agent, partnerID, TPTP_DATA_PATH_DIRECT_SOCKET | TPTP_DATA_PATH_SEND, sourceID, context );
			}
			else
			{
				if ( flags & TPTP_DATA_PATH_MULTIPLEXED )
				{
					ret = handleEstablishDataPathCommand( stateData, agent, partnerID, TPTP_DATA_PATH_SEND | TPTP_DATA_PATH_MULTIPLEXED, sourceID, context );
				}
				else
				{
					ret = handleEstablishDataPathCommand( stateData, agent, partnerID, TPTP_DATA_PATH_SEND, sourceID, context );
				}
			}
		}
		else if ( isEqualString( cmdName, "releaseDataPath" ) )
		{
			int partnerID;

			if (0 != getIntegerParam( "dataConnectionID", paramList, &partnerID ) )
			{
				TPTP_LOG_ERROR_MSG( stateData, "The releaseDataPath command arrived without a dataConnectionID.");
				return -1;
			}

			ret = handleReleaseDataPathCommand( stateData, agent, partnerID );
		}
		else
		{
			/* Log this as an unrecognized command */
			TPTP_LOG_ERROR_MSG2( stateData, "ACTL Received unrecognized command: iid=%s, cmd=%s", interfaceID, cmdName );
			return -1;
		}
	}
	else if ( isEqualString( interfaceID, "org.eclipse.tptp.legacy" ) )
	{
		/* Handle legacy commands that have simply been passed through */
		if ( isEqualString( cmdName, "customCommand" ) )
		{
			ra_message_t* message;
			ra_command_t* command;
			tptp_string*  data = NULL;
			int           dataStrLen;
			int len;
			char *buffer;

			getStringParam( "message", paramList, &data );

			message = ra_createMessage(RA_CONTROL_MESSAGE, 0);
			command = ra_addCommandToMessage(message, NULL);
			command->tag = RA_CUSTOM_COMMAND;
			command->info.custom_command.context = context;
			ra_copyRASTRING( &(command->info.custom_command.agent), &agent->agentName );
			command->info.custom_command.processId = agent->process->pid;
			
			/* Convert the base64 data to binary */
			dataStrLen = strlen(data);

			/* Calculate the size of buffer required */
			len = tptp_decodeBase64( data, dataStrLen, NULL, 0 );

			/* Bug 191075 begins */
			buffer = (char*)tptp_malloc(sizeof(char) * len);
			tptp_decodeBase64(data, dataStrLen, buffer, len);

			if(buffer[len - 1] == 0) { /* Trim the null if found */
				command->info.custom_command.message.data = (char*)tptp_malloc(sizeof(char) * (len - 1));
				memcpy(command->info.custom_command.message.data, buffer, len - 1);
				command->info.custom_command.message.length = len - 1; 
				tptp_free(buffer);
			}
			else {
				command->info.custom_command.message.data = buffer;
				command->info.custom_command.message.length = len; 
			}
			/* Bug 191075 ends */

			if (data != NULL)
			{
				tptp_free(data);
			}

			ra_forwardMessage(message, agent);
			ra_destroyMessage(message, TRUE);

			return TRUE;
		}
		else if ( isEqualString( cmdName, "binaryCustomCommand" ) )
		{
			ra_message_t* message;
			ra_command_t* command;
			tptp_string*  data = NULL;
			int           dataStrLen;
			int len;

			getStringParam( "message", paramList, &data );

			message = ra_createMessage(RA_CONTROL_MESSAGE, 0);
			command = ra_addCommandToMessage(message, NULL);
			command->tag = RA_BINARY_CUSTOM_COMMAND;
			command->info.custom_command.context = context;
			ra_copyRASTRING( &(command->info.custom_command.agent), &agent->agentName );
			command->info.custom_command.processId = agent->process->pid;

			/* Convert the base64 data to binary */
			dataStrLen = strlen(data);

			/* Calculate the size of buffer required */
			len = tptp_decodeBase64( data, dataStrLen, NULL, 0 );
			command->info.custom_command.message.data = (char*)tptp_malloc( sizeof(char) * len );
			tptp_decodeBase64( data, dataStrLen, command->info.custom_command.message.data, len );
			command->info.custom_command.message.length = len;

			if (data != NULL)
			{
				tptp_free(data);
			}

			ra_forwardMessage(message, agent);
			ra_destroyMessage(message, TRUE);

			return TRUE;
		}
		else if ( isEqualString( cmdName, "setNameValuePair" ) )
		{
			ra_message_t* message;
			ra_command_t* command;
			tptp_string*  name  = NULL;
			tptp_string*  type  = NULL;
			tptp_string*  value = NULL;

			getStringParam( "name",  paramList, &name  );
			getStringParam( "type",  paramList, &type  );
			getStringParam( "value", paramList, &value );

			message = ra_createMessage(RA_CONTROL_MESSAGE, 0);
			command = ra_addCommandToMessage(message, NULL);
			command->tag = RA_SET_NAME_VALUE_PAIR;
			command->info.set_nv_pair.context = context;
			ra_copyRASTRING( &(command->info.set_nv_pair.agent), &agent->agentName );

			command->info.set_nv_pair.processId = agent->process->pid;
			ra_createRASTRING( &command->info.set_nv_pair.type,  type );
			ra_createRASTRING( &command->info.set_nv_pair.name,  name );
			ra_createRASTRING( &command->info.set_nv_pair.value, value );

			if (name != NULL)
				tptp_free(name);
			if (type != NULL)
				tptp_free(type);
			if (value != NULL)
				tptp_free(value);

			ra_forwardMessage(message, agent);
			ra_destroyMessage(message, TRUE);

			return TRUE;
		}
		else if ( isEqualString( cmdName, "getTypeAndUUID" ) )
		{
			/* We're intercepting this command to provide deprecated information about the agent */
			char responseFmt[] = "<Cmd src=\"%ld\" dest=\"%ld\" ctxt=\"%ld\"><agentTypeAndUUID iid=\"org.eclipse.tptp.legacy\"><agentUUID>%s</agentUUID><agentType>%s</agentType></agentTypeAndUUID></Cmd>";
			char responseCmd[8192];

			sprintf( responseCmd, responseFmt, agent->agentID, sourceID, context, agent->agentUUID.data, agent->agentType.data );

			forwardXmlCommand( stateData, responseCmd );
		}
		else if ( isEqualString( cmdName, "notifyProcessScrub" ) )
		{
			actl_state_data_t*        actlData  = (actl_state_data_t*)stateData->implData;
			PID                       pid;
			tptp_processScrubEntry_t* scrub;

			if (0 != getPIDParam( "pid", paramList, &pid ) )
			{
				TPTP_LOG_ERROR_MSG( stateData, "ACTL received notifyProcessScrub with no PID" );
				return -1;
			}

			/* Add this process to the scrub list */
			scrub = (tptp_processScrubEntry_t*)tptp_malloc( sizeof(tptp_processScrubEntry_t) );
			if ( scrub != NULL )
			{
				scrub->agent = agent;
				scrub->pid = pid;

				tptp_list_add( &actlData->processScrubList, scrub );

				/* Notify the scrub thread that we've got one waiting */
				tptp_postSemaphore( &actlData->processScrubSem );
			}
		}
		else if ( isEqualString( cmdName, "waitForDataFlush" ) )
		{
			char                          responseFmt[] = "<Cmd src=\"%ld\" dest=\"%ld\" ctxt=\"%ld\"><waitForDataFlushComplete iid=\"org.eclipse.tptp.legacy\"></waitForDataFlushComplete></Cmd>";
			char                          responseCmd[8192];

			if ( agent->isProcessingData )
			{
				agent->dataReceived = 0;
				startFlushingTimeoutThread( agent );
				tptp_waitSemaphore( &agent->dataSemaphore );
				/* The process inactive handle may be waiting too */
				tptp_postSemaphore( &agent->dataSemaphore );
			}

			/* Send a response */
			sprintf( responseCmd, responseFmt, agent->agentID, sourceID, context  );
			forwardXmlCommand( stateData, responseCmd );
		}
		else if ( isEqualString( cmdName, "stopDataFlush" ) )
		{
			tl_data_connection_t*         dataConnection;
			actl_data_connection_block_t* dataInfo;

			/* Get a read lock on the shared memory, so it won't be deleted from under our nose */
			tptp_getReadLock( &stateData->dataConnectionLock );

			dataConnection = (tl_data_connection_t *) tableGet(stateData->dataConnectionTable, agent->dataConnectionID);

			if (dataConnection == NULL)
			{
				TPTP_LOG_ERROR_MSG2( stateData, "ACTL stopDataFlush agent %s invalid dataConnection %d ", agent->agentName.data, agent->dataConnectionID);
				tptp_releaseReadLock( &stateData->dataConnectionLock );
				return -1;
			}

			dataInfo = (actl_data_connection_block_t*) dataConnection->implData;

			//Stop the flushing routine.
			//The flushing thread will cleanup after itself.
			if (dataInfo == NULL)
			{
				TPTP_LOG_ERROR_MSG2( stateData, "ACTL stopDataFlush agent %s invalid shared buffer for dataConnection %d ", agent->agentName.data, agent->dataConnectionID);
				tptp_releaseReadLock( &stateData->dataConnectionLock );
				return -1;
			}
			
			ipcStopFlushing(dataInfo->memBlock);
			
			/* Let go of the shared memory */
			tptp_releaseReadLock( &stateData->dataConnectionLock );
		}
		else if ( isEqualString( cmdName, "errorString" ) )
		{
			ra_message_t* message;
			ra_command_t* command;
			tptp_uint32 errorSeverity = 0;
			tptp_string*  errorMsg = NULL;
			tptp_string*  errorMsgId = NULL;
			int           errorMsgStrLen;
			int           errorMsgIdStrLen;
			int len;

			getIntegerParam("severity", paramList, &errorSeverity);
			getStringParam("messageId", paramList, &errorMsgId);
			getStringParam("message", paramList, &errorMsg);

			message = ra_createMessage(RA_CONTROL_MESSAGE, 0);
			command = ra_addCommandToMessage(message, NULL);
			command->tag = RA_ERROR_STRING;
			command->info.error_string.context = context;
			ra_copyRASTRING( &(command->info.error_string.agent), &agent->agentName );
			command->info.error_string.processId = agent->process->pid;

			/* Convert the base64 message to binary */
			errorMsgIdStrLen = strlen(errorMsgId);

			/* Calculate the size of buffer required */
			len = tptp_decodeBase64(errorMsgId, errorMsgIdStrLen, NULL, 0);
			command->info.error_string.messageId.data = (char*)tptp_malloc(sizeof(char) * len);
			tptp_decodeBase64(errorMsgId, errorMsgIdStrLen, command->info.error_string.messageId.data, len);
			command->info.error_string.messageId.length = len;

			errorMsgStrLen = strlen(errorMsg);

			/* Calculate the size of buffer required */
			len = tptp_decodeBase64(errorMsg, errorMsgStrLen, NULL, 0); 
			command->info.error_string.message.data = (char*)tptp_malloc(sizeof(char) * len);
			tptp_decodeBase64(errorMsg, errorMsgStrLen, command->info.error_string.message.data, len);
			command->info.error_string.message.length = len;

			if (errorMsgId != NULL)
				tptp_free(errorMsgId);
			if (errorMsg != NULL)
				tptp_free(errorMsg);

			ra_forwardMessage(message, agent);
			ra_destroyMessage(message, TRUE);

			return TRUE;
		}
		else
		{
			/* Log this as an unrecognized command */
			TPTP_LOG_ERROR_MSG2( stateData, "ACTL Received unrecognized command: iid=%s, cmd=%s", interfaceID, cmdName );
			return -1;
		}
	}
	else if ( isEqualString( interfaceID, "org.eclipse.tptp.agentManager" ) )
	{
		/* Handle responses to commands we sent to the Agent Manager */
		if ( isEqualString( cmdName, "agentRegistered" ) )
		{
			tptp_list_t propertyList;
			tptp_string* agentConfigData = NULL;

			tptp_list_init( &propertyList );

			getStringListParam( "propertyListEntry", paramList, &propertyList );
			ret = getXmlFragmentParam( "agentConfigData", paramList, &agentConfigData );

			ret = handleAgentRegistered( stateData, agent, &propertyList, agentConfigData );

			tptp_list_clear( &propertyList );
		}
		else if ( isEqualString(cmdName, "agentDeregistered") )
		{
			TPTP_LOG_DEBUG_MSG( stateData, "Received agentDeregistered response" );
		}
		else if ( isEqualString(cmdName, "dataConnectionsBound") )
		{
			ret = handleDataConnectionsBoundCommand( stateData, agent, context );
		}
		else if ( isEqualString( cmdName, "propertyList" ) )
		{
			//we need to reformat this as a response to the client
			ret = handlePropertyList( stateData, agent, context, paramList );
		}
		else if ( isEqualString( cmdName, "agentReference" ) )
		{
			//we need to parse the running agents and then
			//formulate a response to the client.
			//the contextData will tell us how to process.
			ret = handleAgentReference( stateData, agent, context, paramList );
		}
		else
		{
			/* Log this as an unrecognized command */
			TPTP_LOG_ERROR_MSG2( stateData, "ACTL Received unrecognized command: iid=%s, cmd=%s", interfaceID, cmdName );
			return -1;
		}
	}
	else if ( isEqualString( interfaceID, "org.eclipse.tptp.processController" ) ) 
	{
		if ( isEqualString( cmdName, "getProcessUUID" ) )
		{
			ret = handleGetProcessUUID( stateData, agent, context, paramList );
		}
		else
		{
			/* Log this as an unrecognized command */
			TPTP_LOG_ERROR_MSG2( stateData, "Received unrecognized command: iid=%s, cmd=%s", interfaceID, cmdName );
			return -1;
		}
	}
	else if ( isEqualString( interfaceID, "org.eclipse.tptp.ac.peerMonitoring" ) )
	{
		/* Handle responses to commands we sent to the Agent Manager */
		if ( isEqualString( cmdName, "connectionInfo" ) )
		{
			tptp_string* ci;

			ret = getXmlFragmentParam( "sourceConnectionInfo", paramList, &ci );
			if ( ret != 0 )
			{
				TPTP_LOG_ERROR_MSG( stateData, "The connectionInfo command arrived without sourceConnectionInfo.");
				return -1;
			}

			ret = handleConnectionInfoReceived( stateData, agent, ci, context );
		}
		else if ( isEqualString( cmdName, "monitorRequested" ) )
		{
		}
	}
	else if ( isEqualString( interfaceID, TPTP_ERROR_IID ) )
	{
		if ( isEqualString( cmdName, "TPTP_Error" ) )
		{
			generic_context_t *contextStruct;

			ret = baseTL_getContextData( stateData, context, (void**) &contextStruct );
			if ( ret == 0 )
			{
				switch ( contextStruct->contextDataType )
				{
					case REQUEST_MONITOR_CONTEXT_TYPE:
						{
							actl_peer_monitoring_context_t* pmcontext=(actl_peer_monitoring_context_t*)contextStruct;
							ra_message_t *forwardMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
							ra_command_t *command=ra_addCommandToMessage(forwardMessage, 0);
							pmcontext->command->tag=RA_PEER_UNREACHABLE;
							ra_addCommandToMessage(forwardMessage, pmcontext->command);
							TPTP_LOG_DEBUG_MSG(stateData, "Forarding RA_PEER_UNREACHABLE to agent");
							ra_forwardMessage(forwardMessage, pmcontext->sourceAgent);
							ra_destroyMessage(forwardMessage, FALSE);
						}
						break;
				}
			}
		}
	}
	else
    {
		/* Log this as an unrecognized interface */
		TPTP_LOG_ERROR_MSG2( stateData, "ACTL Received command for unrecognized interface: iid=%s, cmd=%s", interfaceID, cmdName );
		return -1;
	}
	
	return 0;
}

/*
 * The thread function of the polling mechanism
 */
#ifdef _WIN32
DWORD WINAPI processPurgeThread(LPVOID args) 
{
#else
void *processPurgeThread(void *args) 
{
#endif
	tl_state_data_t* stateData = (tl_state_data_t*)args;
	actl_state_data_t*  actlData  = (actl_state_data_t*)stateData->implData;

	while(TRUE) 
	{
		/* Wait to be notified that there is a process to scrub */
		tptp_waitSemaphore( &actlData->processScrubSem );

		/* Get the process list lock */
		ra_mutexEnter( &actlData->processList->lock );

		/* While messages are being processed the lock is open but
		   it isn't safe to purge processes.  We'll be signaled again
		   when it's OK to purge */
		if ( !actlData->agentInUse )
		{
			/* Scrub all waiting processes */
			while ( actlData->processScrubList.head != NULL )
			{
				tptp_processScrubEntry_t* scrub = actlData->processScrubList.head->data;
				process_t*                process;

				process=ra_processFind(actlData->processList, scrub->pid);

				/* NULL here is usually OK.  If the process had multiple agents, we'll be notified once
					  for each agent, but the process is destroyed on the first notification */
				if ( process != NULL )
				{
					TPTP_LOG_DEBUG_MSG1( stateData, "ACTL scrubbing process %lu", scrub->pid);
					baseTL_removeControlConnectionEntry( stateData, scrub->agent->agentID );
					ra_processDestroy(process);
				}

				/* This will bring something else (maybe NULL) to the head */
				tptp_list_remove( &actlData->processScrubList, scrub );
			}
		}

		/* Release the lock */
		ra_mutexExit( &actlData->processList->lock );
	}

	return 0;
}



/*
 * Thread to keep polling the registered processes and see if they are still alive
 */
TID startProcessPurgeThread(tl_state_data_t* stateData) 
{
	TID                tid;
	HANDLE			   handle;
	actl_state_data_t* actlData  = (actl_state_data_t*)stateData->implData;

	/* Get our wait mechanisms ready */
	tptp_initializeSemaphore( &actlData->processScrubSem );

	#ifdef MVS
		// processScrubSem has no corresponding delete call, so we register it as one that must be deleted on process shutdown
		tptp_zosRegisterUndeletedSemaphore(&actlData->processScrubSem);
	#endif

	tptp_list_init( &actlData->processScrubList );
	actlData->agentInUse = 0;

#if defined(_WIN32)
	handle = CreateThread(NULL,			/* default security attributes */
						0,						/* same stack size as current thread */
						processPurgeThread,	    /* thread entry point */
						(LPVOID)stateData,		/* thread params */
						0,						/* start executing immediately */
						&tid);					/* the thread ID */
	CLOSE_THREAD_HANDLE(handle);
#elif defined(_AIX)
	pthread_attr_t thread_attr;

	pthread_attr_init(&thread_attr);
	pthread_attr_setstacksize( &thread_attr, 4194304 );
	pthread_create(&tid,
		&thread_attr,
		processPurgeThread,
		(void *)stateData);
#else
	pthread_create(&tid,
		NULL,
		processPurgeThread,
		(void *)stateData);
#endif
	TPTP_LOG_DEBUG_MSG1( stateData, "ACTL process purge thread created, tid = %d", tid );

	return tid;
}


