/**********************************************************************
 * Copyright (c) 2005, 2009 IBM 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
 * $Id: RABindings.c,v 1.8 2009/08/12 19:45:47 jwest Exp $
 *
 * Contributors:
 * IBM - Initial API and implementation
 **********************************************************************/

#include <stdio.h>
#include <time.h>
#include <stdlib.h>
#ifndef __OS400__
	#include <sys/timeb.h>
#endif
#include <sys/types.h>

/** Platform specifics */
#ifdef _WIN32								/* _WIN32 */
  #define TIMEB  struct _timeb
  #define FTIME(param)  _ftime(param)
#else										/* not Win32 */
  #include <sys/socket.h>
  #include <netinet/in.h>
  #include <netdb.h>
  #include <unistd.h>
  #include <pthread.h>
  #include <errno.h>
  #ifdef _HPUX
		#include <stdlib.h> /* for atexit() */
		#include <ctype.h> /* for isdigit() */
  #endif
  #define TIMEB	struct timeb
  #define FTIME(param)  ftime(param)
#endif
										/* endif */
#include "lock3.h"
#include "RAError.h"
#include "RASocket.h"
#include "RABindings.h"

#define AGENT_START_PORT 10004
#define MESSAGE_BUFFER_INCREMENT  1024
#define MAX_AGENT 512 /* maximum number of agents allowed */

/**
  * Globals
  */
#ifdef _WIN32
static HANDLE _vmLock=NULL;
#else
static pthread_mutex_t  _vmLock_mutex;
static pthread_cond_t   _vmLock_cond;
#endif
static BOOL   _pendingVMLock;

static BOOL _serverInitialized=FALSE;


static unsigned long _usageCount=0;

#ifdef _DEBUG
static FILE *dbgFile=NULL;
#endif

/** LinuxThreads follows the so-called "one-to-one" model: each thread is actually
    a separate process in the kernel. The kernel scheduler takes care of scheduling
    the threads, just like it schedules regular processes. The threads are created
    with the Linux clone() system call, which is a generalization of fork() allowing
    the new process to share the memory space, file descriptors, and signal handlers
    of the parent.  In the end this maps to a different result when calling getpid()
    when in different threads.  Providing startListener is invoked on the main thread
    of the process I will cache the process ID here and retrieve it using
    ra_getCurrentProcessId()
  */
#if defined __linux__ || defined _HPUX
static PID _pid;
#endif

/**
  * Function prototypes
  */
static BOOL processControlMessage(RA_AGENT_HANDLE handle,
                                  ra_message_t *message); 

static int readConnection(RA_HANDLE connection,
                          unsigned char *buffer,
                          int offset,
                          int length,
                          unsigned int *bytesRead);

static void messageLoop(RA_AGENT_HANDLE agentHandle, int timeout); /* APAR PK01709: Configurable named pipe timeout */

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

static ra_uint_t errorRecoveryScan(unsigned char *buffer,
                                   int length);

static ra_uint_t getMessageLengthfromValidBuffer(unsigned char *buffer);

static void sendAgentInactive(RA_AGENT_HANDLE handle);

void pipeServerCleanup(void *arg);


void *PipeServer(void *args);
#ifdef _WIN32
DWORD WINAPI win32pipeServerProxy(LPVOID args);
#endif

/* List and function to store agents for cleaning up */
int agentHandleCount = 0;
RA_AGENT_HANDLE agentHandleList[MAX_AGENT];
static void addHandle(RA_AGENT_HANDLE handle);
static void removeAllHandles();
static void removeHandle(RA_AGENT_HANDLE handle);

/** INITIALIZE_BINDINGS  ***********************************************
  * Initialization routine for the agent bindings, must be called first
  * by te agent before running any of the other functions specified in this
  * header file.
  * @param       name - the name of the agent.
  * @param       type - the type of the agent
  * @param    handler - the callback routine for all incomming messages.
  * @param standalone - determines whether this agent should communicate with the RAC
  *                     if standalone is TRUE the RAC is never used.
  * @returns  the new handle that is used to manage the agent.
  */
RA_AGENT_HANDLE ra_initializeBindings(const char *name,
                                      const char *type,
                                      ra_commandHandler_t handler,
                                      BOOL standalone) {

	/* Allocate the handle */
	RA_AGENT_HANDLE handle=(RA_AGENT_HANDLE)ra_malloc(sizeof(ra_agenthandle_t));
	BZERO(handle, sizeof(ra_agenthandle_t));

	/* At this point we are neither registered with the RAC nor are our bindings finalized */
	handle->registered=handle->finalized=FALSE;
	handle->pipefd = -1; /* 46651 */

	/* Setup the command handler */
	handle->commandHandler=handler;

	/* Generate a UUID for this agent */
	ra_generateUUID(&handle->agentUUID);

	/* Create the agent name/type */
	ra_createRASTRING(&handle->agentName, name);
	ra_createRASTRING(&handle->agentType, type);

	/* We won't know this information until we register with the RAC */
	handle->nodeUUID.length=0;
	handle->nodeUUID.data=NULL;
	handle->processUUID.length=0;
	handle->processUUID.data=NULL;

	/* TO_DO:  Determine node UUID in standalone mode for now..generate anew every time */
	if(!handle->nodeUUID.length) {
		ra_generateUUID(&handle->nodeUUID);
	}

	if(standalone && !handle->processUUID.length) {
		ra_generateUUID(&handle->processUUID);
	}

	#ifdef _WIN32
		// Bug 278655 - We will keep track of whether the PipeServer() thread has yet terminated.
		handle->pipeServerThreadEnded = 0;
	#endif

	/* The first agent needs to start the RAC watcher as well as cache the process ID on Linux */
	if(!_serverInitialized) {
		atexit(removeAllHandles);
/* BEGIN: 235649 */
#if defined __linux__ || defined _HPUX
		/* Save the process ID for later retrieval */
		if(!_pid) {
			_pid=getpid();
		}
#endif
/* END: 235649 */
		_serverInitialized=TRUE;
	}


	/* Copy the process ID into the storage structure */
	handle->processID=ra_getProcessId();


	/* Set the default configuration to be empty */
	handle->defaultConfiguration.head=handle->defaultConfiguration.tail=NULL;

	_usageCount++;

	return handle;
}

/** FINALIZE_BINDINGS  *********************************************************
  * Cleans up all the resources that were allocated during the agents
  * interaction with the bindings. Once this is called the handle is
  * no longer valid.
  * @param  handle - a handle returned by ra_initializeBindings.
  */
extern void ra_finalizeBindings(RA_AGENT_HANDLE handle) {
	ra_destroyRASTRING(&handle->agentUUID);
	ra_destroyRASTRING(&handle->nodeUUID);
	ra_destroyRASTRING(&handle->processUUID);
	ra_destroyRASTRING(&handle->agentName);
	ra_destroyRASTRING(&handle->agentType);
	_usageCount--;
}


/** START_LISTENER  ************************************************************
  * Start the message processor.
  * @param   handle - a handle returned by ra_initializeBindings
  * @param    block - whether to  block until the agent is successfuly
  *                   attached to a remote client.
  * @returns    NULL  call failed, call ra_getLastError() to determine failure
  *		    otherwise the address of the thread number the server message processor is
  *	                  listening on
  *
  */
TID *ra_startListener(RA_AGENT_HANDLE handle,
                      char lock) {

	/* Add the handler code when exiting */
	addHandle(handle);

	/* Create our named pipe */
	handle->pipefd=ra_createNamedPipe(handle->agentUUID.data);

	/* Run the server on a separate thread and return the tid */
#ifdef _WIN32
	handle->tHandle = CreateThread(NULL,					/* default security attributes */
				 0,						/* same stack size as current thread */
				 win32pipeServerProxy,	/* Thread entry point */
				 (LPVOID)handle,		/* params */
				 0,						/* start executing immediately */
				 &(handle->tid));		/* the thread ID */
#else
	errno=0;
#ifdef _AIX
	pthread_attr_t thread_attr;

	pthread_attr_init(&thread_attr);
	pthread_attr_setstacksize( &thread_attr, 4194304 );
	int result = pthread_create(&(handle->tid),
			  &thread_attr,
			  PipeServer,
			  handle);
#else
	int result = pthread_create(&(handle->tid),
			  NULL,
			  PipeServer,
			  handle);
#endif

	if(result!=0) {
		/* bug 64225 begin */
		int thrdCreateLoopIndex = 1;
		int createRC = -1;

		/* Loop a finite number of times trying to create the thread if the operation
		 * fails with errno=EAGAIN
		 */
		while ( createRC != 0 && errno == EAGAIN && thrdCreateLoopIndex < 8 ) {
			#ifdef _DEBUG
			fprintf(dbgFile, "ra_startListener: EAGAIN failure creating PipeServer thread %d times\n", thrdCreateLoopIndex);
			fflush(dbgFile);
			#endif
			sleep(1);
#ifdef AIX
			pthread_attr_t thread_attr;

			pthread_attr_init(&thread_attr);
			pthread_attr_setstacksize( &thread_attr, 4194304 );
			createRC = pthread_create(&(handle->tid),
					  &thread_attr,
					  PipeServer,
					  handle);
#else
			createRC = pthread_create(&(handle->tid),
					  NULL,
					  PipeServer,
					  handle);
#endif
			thrdCreateLoopIndex++;

		}

		/* If a thread still can't be created then quit */
		if (createRC != 0) {
			ra_setLastError(THREAD_CREATE_FAILED, errno);
			#ifdef _DEBUG
			fprintf(dbgFile, "ra_startListener: Failure creating PipeServer thread - errno=%d >%s\n", errno, strerror(errno));
			fflush(dbgFile);
			#endif
			return (TID*)(-1);
		}
		/* bug 64225 end */
	}
#endif

/*	jvmpiAgent_initializeLock(handle); */
	ra_initializeVMLock();


	/* If we are told to block until JVMPI_RESUME_VM */
	if(lock) {
		ra_setVMLock();
		ra_testVMLock();

/*
		jvmpiAgent_getWriteLock(handle);

		/* Don't return until specified
		jvmpiAgent_getWriteLock(handle);
		jvmpiAgent_releaseWriteLock(handle);
*/
	}
	return &handle->tid;
}

/** STOP_LISTENER  *************************************************************
  * Stops the message processor
  * @param  handle - a handle returned by ra_initializeBindings
  */
void ra_stopListener(RA_AGENT_HANDLE handle) {

#ifndef _WIN32
	void *status;
#endif
	int waitResult = 0;
	/* Set that this handle is no longer valid.  The PipeServer created in initializeBindings
	 * will then exit and cleanup the remainder of this agents info.
	 */

	/* AS/400 need to call this because the thread_cleanup_push() isn't working as expected.
	 *	We may need to revisit this thread issue on the AS/400
	 */
#ifdef _DEBUG
	fprintf(dbgFile, "ra_stopListener: about to stop listener thread\n");
	fflush(dbgFile);
#endif

	handle->finalized=TRUE;
#if defined _WIN32
	/* we need to do this on Windows so that the PipeServer doesn't block on any 
	   calls dealing with the agent's pipe  -- we do this on other platforms when
	   pipeServerCleanup is called */ 
	ra_closeNamedPipe(handle->pipefd); 

	// Bug 278655 - 20 seconds is an arbitrary value
	waitResult = WaitForSingleObject(handle->tHandle, 20000);

	// Bug 278655 - It is unsafe to continue from this point if the PipeServer() thread has not yet completed, so use WFSO
	if(waitResult == WAIT_TIMEOUT && !(handle->pipeServerThreadEnded)) {
		waitResult = WaitForSingleObject(handle->tHandle, INFINITE);
	}

	CloseHandle(handle->tHandle); 
#endif
#if defined _WIN32 || defined __OS400__ /* 238529 */
	pipeServerCleanup(handle);
#endif


	/* On posix we need to cancel the message processing thread so
	 * it can cleanup its resources.
    * 227339 - we need to wait until the message processing thread
    * ends to ensure the cleanup happens
	 */
#ifndef _WIN32
	// Bug 284438 - If this thread issues a cancel on the PipeServer thread, and then exits
	// before that thread is complete (e.g. does not join it), it causes a segfault in the JVM (likely a JVM bug). However, as
	// implemented, the PipeServer thread will not cancel in certain cases, leading
	// the JVM to hang around forever. As a workaround I am removing the cancel and join call,
	// which, while not perfect, seems acceptable.

	// pthread_cancel(handle->tid);
	// pthread_join(handle->tid, &status);
#endif

}

/** SEND_MESSAGE  **************************************************************
  * Send the message to the RAC on this node.  It will
  * then be forwarded by the server to the client.
  * @param  handle - a handle returned by ra_initializeBindings
  * @param   message - the message structure to send.
  * @returns the number of bytes sent to the RAC.
  */
int ra_sendMessage(RA_AGENT_HANDLE handle,
                   ra_message_t *message) {
	int byte=0, length=0;
	char readyBuffer[MESSAGE_BUFFER_INCREMENT];
	char *buffer=NULL;
	int result;
	int paddingGuess = 64; /* we have to guess the number of padding byes for all RASTRING */	
	RA_HANDLE racConnection = -1; /* 46651 */
	int retryAttempts = 600; /* Bug 146962 */ /* number of times to retry our connection to the master pipe */ 

	BOOL bufferMalloced = FALSE;

	/* Reconnect the server pipe */
	while ( (result=ra_openNamedPipe(RA_MASTER_ADDRESS)) < 0 && !handle->finalized && retryAttempts-- > 0) {
		SLEEP(100); /* sleep before connect reattempt so as not to create tightly spinning loop
					   that consumes CPU */ 
	}

	if (result < 0) {
		return 0;
	}

	racConnection = result; 
	/* Get the length here as we could overflow, otherwise use the static buffer */
	length=ra_determineMessageLength(message);

	#ifdef _DEBUG
	fprintf(dbgFile, "ra_sendMessage: about to write message of length %d\n", length);
	fflush(dbgFile);
	#endif

	if(length>MESSAGE_BUFFER_INCREMENT) {
		length=length*sizeof(char) + paddingGuess;
		buffer=(char*)ra_malloc(length); /* allocate with the extra padding bytes */			
		bufferMalloced = TRUE;
	}
	else {
		buffer=readyBuffer;
		length = MESSAGE_BUFFER_INCREMENT;
	}

	#ifdef _DEBUG
	fprintf(dbgFile, "ra_sendMessage: about to write message to buffer %x  with length %d\n", buffer, length);
	fflush(dbgFile);
	#endif

	/* Write the message to the buffer */
	length=ra_writeMessageToBuffer((unsigned char *)buffer, length, message);

	#ifdef _DEBUG
	fprintf(dbgFile, "ra_sendMessage: about to write message of length %d to named pipe\n", length);
	fflush(dbgFile);
	#endif

	/* Send to the RAC */
	if(racConnection >= 0) { /* 46651 */
		result=ra_writeToNamedPipe(racConnection, buffer, 0, length, &byte);
		ra_closeNamedPipe(racConnection);
		#ifdef _DEBUG
		fprintf(dbgFile, "ra_sendMessage: wrote %d bytes to named pipe - result=%d\n", byte, result);
		fflush(dbgFile);
		#endif
	}

	if(bufferMalloced) {
		ra_free(buffer);
	}

	return byte;
}

/** GET_PROCESS_ID **************************************************
  * Multiplatform convenience method to retrieve the current process
  * id.
  * @returns the PID of the curren process
  */
PID ra_getProcessId() {
#ifdef _WIN32
	return GetCurrentProcessId();
#elif defined __linux__ || defined _HPUX
	return _pid;
#else
	return getpid();
#endif
}

/** GET_NODE_UUID  *****************************************************
  * Retrieves the UUID of the node which this agent is running on.
  * Calling this function before ra_startListener results in undefined
  * behaviour.
  */
const char* ra_getNodeUUID(RA_AGENT_HANDLE storage) {
	if(storage->registered) {
		return storage->nodeUUID.data;
	}
	return NULL;
}

/** GET_PROCESS_ID  ****************************************************
  * Retrieves the UUID of the process which this agent is running in.
  * Calling this function before ra_startListener results in undefined
  * behaviour.
  */
const char* ra_getProcessUUID(RA_AGENT_HANDLE storage) {
	if(storage->registered) {
		return storage->processUUID.data;
	}
	return NULL;
}

/** GET_AGENT_UUID  ************************************************************
  * Retrieves the agent's UUID.
  */
const char* ra_getAgentUUID(RA_AGENT_HANDLE storage) {
	return storage->agentUUID.data;
}

/** GET_DEFAULT_CONFIGURATION  *************************************************
  * Retrieves a list of the options provided by the RAC as a default configuration
  * of the agent.  The configuration will only be available after the agent
  * has successfuly registered with the RAC.
  * @param  handle - a handle returned by ra_initializeBindings
  * @returns   the agent's default configuration list if the agent has registered
  *            with the RAC.  NULL otherwise.
  */
ra_agentConfigList_t* ra_getDefaultConfiguration(RA_AGENT_HANDLE handle) {
	if(!handle->registered) {
		return NULL;
	}
	return &handle->defaultConfiguration;
}


/** GET_PROCESS_COMMAND_LINE  ******************************************
  */
char* ra_getProcessCommandLine() {
#ifdef _WIN32
	char *temp;
	char * cmdLine=GetCommandLine();
	temp=cmdLine;
	while ((temp=strchr(temp, '\"'))!=NULL) {
		*temp=' ';
	}
	return cmdLine;
#else
	return NULL;
#endif
}

/** LOG_ERROR_MESSAGE  **************************************************
  *
  */
/* Bug 74367 */
int ra_logErrorMessage_return(RA_AGENT_HANDLE storage, RA_ERROR_LEVEL  severity, const char *messageId, const char *message) {
	int byteSent = 0;
	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=0;
	errorCommand->info.error_string.processId=ra_getProcessId();
	ra_copyRASTRING(&errorCommand->info.error_string.agent, &storage->agentName);
	errorCommand->info.error_string.severity=severity;
	ra_createRASTRING(&errorCommand->info.error_string.messageId, messageId);
	ra_createRASTRING(&errorCommand->info.error_string.message, message);
	byteSent = ra_sendMessage(storage, errorMessage);
	ra_destroyMessage(errorMessage, TRUE);
	return byteSent;
}

void ra_logErrorMessage(RA_AGENT_HANDLE storage, RA_ERROR_LEVEL  severity, const char *messageId, const char *message) {
	ra_logErrorMessage_return(storage, severity, messageId, message);
}


/** REQUEST_PEER_MONITOR  ******************************************************
  * Requests a remote client to start monitoring this agent by asking the agent
  * controller to find the client currently monitoring a peer agent.
  * THIS IS NOT CURRENTLY SUPPORTED.  DO NOT USE!!!!!!!
  */
extern int ra_requestPeerMonitor(RA_AGENT_HANDLE storage, ra_ipaddr_t *peerAddress, PID peerPID, ra_string_t *peerAgent, unsigned long timeout) {
	return ra_requestPeerMonitor_p(storage, peerAddress, 10002, peerPID, peerAgent, timeout);
}

extern int ra_requestPeerMonitor_p(RA_AGENT_HANDLE storage, ra_ipaddr_t *peerAddress, int peerPort, PID peerPID, ra_string_t *peerAgent, unsigned long timeout) {

	ra_message_t *attachRequestMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
	ra_command_t *command=ra_addCommandToMessage(attachRequestMessage, NULL);

	if(peerPort == RA_CTL_PORT_NUM_SERVER) {
		command->tag=RA_AGENT_REQUEST_MONITOR;
		command->info.agent_request_monitor.context=0;
		command->info.agent_request_monitor.processId=ra_getProcessId();
		ra_copyRASTRING(&command->info.agent_request_monitor.agent, &storage->agentName);
		ra_createRASTRING(&command->info.agent_request_monitor.node, NULL);	/* This will be filled in by the RAC */

		/* Peer info */
		ra_createRASTRING3(&command->info.agent_request_monitor.peerNode, (const char*)peerAddress->addr, peerAddress->addrLength);
		command->info.agent_request_monitor.peerProcessId=peerPID;
		ra_copyRASTRING(&command->info.agent_request_monitor.peerAgent, peerAgent);
	}
	/* Bug 77768 */
	else {
		command->tag=RA_AGENT_REQUEST_MONITOR_PORT;
		command->info.agent_request_monitor_port.context=0;
		command->info.agent_request_monitor_port.processId=ra_getProcessId();
		ra_copyRASTRING(&command->info.agent_request_monitor_port.agent, &storage->agentName);
		ra_createRASTRING(&command->info.agent_request_monitor_port.node, NULL);	/* This will be filled in by the RAC */

		/* Peer info */
		ra_createRASTRING3(&command->info.agent_request_monitor_port.peerNode, (const char*)peerAddress->addr, peerAddress->addrLength);
		command->info.agent_request_monitor_port.peerProcessId=peerPID;
		ra_copyRASTRING(&command->info.agent_request_monitor_port.peerAgent, peerAgent);
		command->info.agent_request_monitor_port.port = 0;
		command->info.agent_request_monitor_port.peerPort = peerPort;
	}

	ra_sendMessage(storage, attachRequestMessage);
	ra_destroyMessage(attachRequestMessage, TRUE);
	return 0;
}

/** REQUEST_PEER_MONITOR  ******************************************************
  * Requests a remote client to start monitoring this agent by asking the agent
  * controller to find the client currently monitoring a peer agent.
  */
extern int ra_requestPeerMonitor2(RA_AGENT_HANDLE handle, ra_ipaddr_t *peerAddress, ra_string_t *peerAgentUUID, unsigned long timeout) {
	return ra_requestPeerMonitor2_p(handle, peerAddress, 10002, peerAgentUUID, timeout);
}

extern int ra_requestPeerMonitor2_p(RA_AGENT_HANDLE handle, ra_ipaddr_t *peerAddress, int peerPort, ra_string_t *peerAgentUUID, unsigned long timeout) {

	ra_message_t *attachRequestMessage=ra_createMessage(RA_CONTROL_MESSAGE, 0);
	ra_command_t *command=ra_addCommandToMessage(attachRequestMessage, NULL);

	if(peerPort == RA_CTL_PORT_NUM_SERVER) {
		command->tag=RA_AGENT_REQUEST_MONITOR;
		command->info.agent_request_monitor.context=0;
		command->info.agent_request_monitor.processId=ra_getProcessId();
		ra_copyRASTRING(&command->info.agent_request_monitor.agent, &handle->agentName);
		ra_createRASTRING(&command->info.agent_request_monitor.node, NULL);	/* This will be filled in by the RAC */

		/* Peer info */
		ra_createRASTRING3(&command->info.agent_request_monitor.peerNode, (const char*)peerAddress->addr, peerAddress->addrLength);
		command->info.agent_request_monitor.peerProcessId=0;
		ra_copyRASTRING(&command->info.agent_request_monitor.peerAgent, peerAgentUUID);
	}
	/* Bug 77768 */
	else {
		command->tag=RA_AGENT_REQUEST_MONITOR_PORT;
		command->info.agent_request_monitor_port.context=0;
		command->info.agent_request_monitor_port.processId=ra_getProcessId();
		ra_copyRASTRING(&command->info.agent_request_monitor_port.agent, &handle->agentName);
		ra_createRASTRING(&command->info.agent_request_monitor_port.node, NULL);	/* This will be filled in by the RAC */

		/* Peer info */
		ra_createRASTRING3(&command->info.agent_request_monitor_port.peerNode, (const char*)peerAddress->addr, peerAddress->addrLength);
		command->info.agent_request_monitor_port.peerProcessId=0;
		ra_copyRASTRING(&command->info.agent_request_monitor_port.peerAgent, peerAgentUUID);
		command->info.agent_request_monitor_port.port = 0;
		command->info.agent_request_monitor_port.peerPort = peerPort;
	}

	ra_sendMessage(handle, attachRequestMessage);
	ra_destroyMessage(attachRequestMessage, TRUE);
	return 0;
}



/** PROCESS_CONTROL_MESSAGE  ****************************************
  * Called by processMessage for JVMPI_CTL_MSG messages.
  * Breaks the message up into individual commands and forwards each
  * command to the registered _jvmpicomm_notify_message method.
  */
static BOOL processControlMessage(RA_AGENT_HANDLE handle,
								  ra_message_t *message) { 
	ra_command_list_node_t *current;
	ra_command_t connectedCommand;
	/* Process each command in turn */
	current=message->commands.head;

	while(current != NULL ) {
		/* Intercept the configuration information */
		if(current->command->tag==RA_AGENT_CONFIGURATION) {
			unsigned int i;
			ra_agentConfigListEntry_t *iterator;

			#ifdef _DEBUG
			fprintf(dbgFile, "processControlMessage: about to process RA_AGENT_CONFIGURATION\n");
			fflush(dbgFile);
			#endif

			/* Cleanup any configuration data from a previous handshake with the RAC */
			ra_destroyRASTRING(&handle->nodeUUID);
			ra_destroyRASTRING(&handle->processUUID);
			/* ra_destroyRASTRING(&handle->nodeUUID); */ 
			iterator=handle->defaultConfiguration.head;
			while(iterator) {
				ra_agentConfigListEntry_t *current;
				current=iterator;
				ra_destroyRASTRING(&current->entry.type);
				ra_destroyRASTRING(&current->entry.name);
				ra_destroyRASTRING(&current->entry.value);
				iterator=current->next;
				ra_free(current);
			}
			handle->defaultConfiguration.head=handle->defaultConfiguration.tail=NULL;


			/* Copy the binding information */
			ra_copyRASTRING(&handle->nodeUUID, &current->command->info.agent_configuration.nodeUUID);
			ra_copyRASTRING(&handle->processUUID, &current->command->info.agent_configuration.processUUID);

			/* Loop through the configuration and create a default configuration for the agent */
			#ifdef _DEBUG
			fprintf(dbgFile, "processControlMessage: Loop through the configuration and create a default configuration for the agent\n");
			fflush(dbgFile);
			#endif
			for(i=0; i<current->command->info.agent_configuration.configuration.length; i++) {
				ra_agentConfigListEntry_t  *copy;
				ra_agentConfigEntry_t *entry;

				entry=((ra_agentConfigEntry_t*)(current->command->info.agent_configuration.configuration.data[i]));
				copy=(ra_agentConfigListEntry_t*)ra_malloc(sizeof(ra_agentConfigListEntry_t));
				ra_copyRASTRING(&copy->entry.type, &entry->type);
				ra_copyRASTRING(&copy->entry.name, &entry->name);
				ra_copyRASTRING(&copy->entry.value, &entry->value);

				#ifdef _DEBUG
				fprintf(dbgFile, "processControlMessage:type=%s\n", copy->entry.type.data);
				fprintf(dbgFile, "processControlMessage:name=%s\n", copy->entry.name.data);
				fprintf(dbgFile, "processControlMessage:value=%s\n", copy->entry.value.data);
				fflush(dbgFile);
				#endif
				/* Insert into the list */
				copy->next=NULL;
				copy->previous=handle->defaultConfiguration.tail;
				handle->defaultConfiguration.tail=copy;

				/* If this not the first entry we must set the link in the previous tail */
				if(copy->previous) {
					copy->previous->next=copy;
				}
				else {
					handle->defaultConfiguration.head=copy;
				}
			}
			/* bugzilla 71840 start */ 
			#ifdef _DEBUG
			fprintf(dbgFile, "processControlMessage: let agent process RA_AGENT_CONTROLER_AVAILABLE\n");
			fflush(dbgFile);
			#endif

			/* Inform the agent we are connected */
			handle->registered=TRUE;
			connectedCommand.tag=RA_AGENT_CONTROLER_AVAILABLE;
			ra_copyRASTRING(&(connectedCommand.info.agentName), &handle->agentName);
			(handle->commandHandler)(&connectedCommand);
		
			#ifdef _DEBUG
			fprintf(dbgFile, "processControlMessage: after handling RA_AGENT_CONTROLER_AVAILABLE command\n");
			fflush(dbgFile);
			#endif

			ra_destroyRASTRING((ra_string_t*)&connectedCommand.info.agentName);
			/* bugzilla 71840 end */ 
			

		}
		else if(current->command->tag==RA_AGENT_CONTROLER_UNAVAILABLE) {
			return FALSE;
		}
		else {

			#ifdef _DEBUG
			fprintf(dbgFile, "procesControlMessage: about to pass command %d to command handler\n", current->command->tag);
			fflush(dbgFile);
			#endif

			/* Call the user defined message handler */
			if(handle->commandHandler) {
				(*handle->commandHandler)(current->command);
			}
		}
		current=current->next;
	}

	return TRUE;
}


void pipeServerCleanup(void *arg) {
	#ifdef _WIN32
		RA_AGENT_HANDLE handle=(RA_AGENT_HANDLE)arg;
	
		sendAgentInactive(handle); 

		removeHandle(handle);
		ra_destroyNamedPipe(handle->agentUUID.data);
		removeHandle(handle);
		ra_free(handle);
	#else
		// Bug 284438 - As a workaround to the issued described in the bug, we allow 
		// the data time to complete before the process completes (note that this function 
		// is called by an atexit() or a an explicit deregister call, which necessarily 
		// requires that all data has already been processed). The rest of the calls
		// removeHandle/ra_free/ra_destroyNamedPipe() are not required, for similar reasons.
		sleep(25);
	#endif
}


/** PIPE_SERVER  ***************************************************************
  * This is the message loop for communications with the RAC.
  */
void *PipeServer(void *args) {
	RA_HANDLE handshakeConnection;

	RA_AGENT_HANDLE handle;
	/* APAR PK01709: Configurable named pipe timeout */
	int pipe_timeout = 1000; /* 1 second default timeout for opening master named pipe */
	char *pipe_timeout_name = "RA_IO_IDLE_TIME";
	char *pipe_timeout_value;
	unsigned int i;
	int rc;
	BOOL isValidTimeout = TRUE;

	/* Determine the pipe timeout */
#ifdef _WIN32
	pipe_timeout_value = (char*)malloc(sizeof(char) * 32);
	BZERO(pipe_timeout_value, 32);
	rc = GetEnvironmentVariable(pipe_timeout_name, pipe_timeout_value, 32);
	if(rc == 0) {
		/* Failed to retrieve the env var */
		pipe_timeout = 1000;
	}
	else if(rc == 16) {
		/* Env var string too long */
		pipe_timeout = 1000;
	}
	else {
		/* Check to see if it is all digit */
		for(i = 0; isValidTimeout && (i < strlen(pipe_timeout_value)); i++) {
			if(!isdigit(pipe_timeout_value[i])) {
				isValidTimeout = FALSE;
			}
		}
		if(isValidTimeout) {
			pipe_timeout = atoi(pipe_timeout_value);
		}
	}
	free(pipe_timeout_value);
#else
	/*printf("DEBUG: Timeout initially set to %d ms\n", pipe_timeout);*/
	pipe_timeout_value = (char*)getenv(pipe_timeout_name);
	if(pipe_timeout_value == NULL) {
		/* Failed to retrieve the env var */
	/*printf("DEBUG: Error retrieving timeout environment, leaving it as %d ms\n", pipe_timeout);*/
		pipe_timeout = 1000;
	}
	else {
	/*printf("DEBUG: Timeout environment variable %s set to: %s\n", pipe_timeout_name, pipe_timeout_value);*/
		/* Check to see if it is all digit */
		for(i = 0; isValidTimeout && (i < strlen(pipe_timeout_value)); i++) {
			if(!isdigit(pipe_timeout_value[i])) {
				isValidTimeout = FALSE;
			}
		}
		if(isValidTimeout) {
			pipe_timeout = atoi(pipe_timeout_value);
		}
		else {
			/*printf("DEBUG: Environment string cannot be converted to an integer value\n");*/
		}
	}
	/*printf("DEBUG: New pipe timeout set to %d ms\n", pipe_timeout);*/
#endif

	handle=((RA_AGENT_HANDLE)args);

	/* On posix we need to push a cleanup handler to ensure we destroy our message
	 * processing thread correctly
	 */
#ifndef _WIN32
	pthread_cleanup_push(pipeServerCleanup, handle);
#endif

	#ifdef _DEBUG
	dbgFile = fopen("hcbnd.dbg", "a");
	fprintf(dbgFile, "==========================PipeServer: starting server===================\n");
	fflush(dbgFile);
	#endif

outer:
	/* Loop as long as our handle is not finalized */
	while(!handle->finalized) {
		BOOL firstConnectTial=TRUE;
		ra_message_t *bindingMessage;
		ra_command_t *bindingCommand;
		unsigned char buffer[1024];
		int bytesWrote;
		unsigned int length;
		int result;
		BOOL sleepPrinted = FALSE; /* APAR PK01709: Configurable named pipe timeout */
		ra_command_t connectedCommand;

		/* Handshake with the RAC */
		firstConnectTial=TRUE;
		#ifdef _DEBUG
		fprintf(dbgFile, "PiprServer: Try to connect to RAC master named pipe\n");
		fflush(dbgFile);
		#endif
		do {
			handshakeConnection=ra_openNamedPipe(RA_MASTER_ADDRESS);
			if((int)handshakeConnection < 0) { /* 46651 */
				if(firstConnectTial) {
					/* Inform the agent that the RAC is not available */
					handle->registered=FALSE;
					connectedCommand.tag=RA_AGENT_CONTROLER_UNAVAILABLE;
					ra_copyRASTRING(&connectedCommand.info.agentName, &handle->agentName);
					(handle->commandHandler)(&connectedCommand);
					ra_destroyRASTRING((ra_string_t*)&connectedCommand.info.agentName);
					firstConnectTial=FALSE;
				}
				/* APAR PK01709: Configurable named pipe timeout */
				if(!sleepPrinted) {
#if defined(MVS)
					fprintf(stdout, "Agent Controller is not active: will wait in a %d ms loop\n", pipe_timeout);
					fflush(stdout);
#else
					printf("Agent Controller is not active: will wait in a %d ms loop\n", pipe_timeout);
#endif
					sleepPrinted = TRUE;
					#ifdef _DEBUG
					fprintf(dbgFile, "PipeServer: Agent Controller is not active: will wait in a %d ms loop\n", pipe_timeout);
					fflush(dbgFile);
					#endif
				}
				SLEEP(pipe_timeout);
			}
			if(handle->finalized){
				goto outer;
			}
		}while((int)handshakeConnection < 0); /* 46651 */ /* Bug 66383 */
		#ifdef _DEBUG
		fprintf(dbgFile, "PipeServer: Connected to RAC master named pipe\n");
		fflush(dbgFile);
		#endif

		#ifdef _DEBUG
		fprintf(dbgFile, "PipeServer: Trying to sent agent registration to the RAC\n");
		fflush(dbgFile);
		#endif

		/* Create our Binding message */
		bindingMessage=ra_createMessage(RA_CONTROL_MESSAGE, 1);
		bindingCommand=ra_addCommandToMessage(bindingMessage, NULL);
		bindingCommand->tag=RA_AGENT_SCOPING_INFORMATION;
		ra_copyRASTRING(&bindingCommand->info.agent_scoping_information.agent, &handle->agentName);
		ra_copyRASTRING(&bindingCommand->info.agent_scoping_information.agentUUID, &handle->agentUUID);
		ra_copyRASTRING(&bindingCommand->info.agent_scoping_information.nodeUUID, &handle->nodeUUID);
		ra_copyRASTRING(&bindingCommand->info.agent_scoping_information.agentType, &handle->agentType);
		ra_copyRASTRING(&bindingCommand->info.agent_scoping_information.processUUID, &handle->processUUID);
		bindingCommand->info.agent_scoping_information.processId=ra_getProcessId();
/* BEGIN:  235649 */
#if defined __linux__
		bindingCommand->info.agent_scoping_information.messageProcessId=getpid();
#endif
/* END: 235649 */
		bindingCommand->info.agent_scoping_information.context=0;

		/* We are handshaking, send our agent binding information */
		ra_determineMessageLength(bindingMessage); /* Bug 129050 */
		length=ra_writeMessageToBuffer(buffer, 1024, bindingMessage);

		/* RKD:  Need to check our return code here */
		result=ra_writeToNamedPipe(handshakeConnection, (char*)buffer, 0, length, &bytesWrote);
		ra_destroyMessage(bindingMessage, TRUE);

		/* Finish the handshake */
		ra_closeNamedPipe(handshakeConnection);

		#ifdef _DEBUG
		fprintf(dbgFile, "PipeServer: Successfully sent agent registration to the RAC\n");
		fflush(dbgFile);
		#endif

		/* APAR PK01709: Configurable named pipe timeout */
		#ifdef _DEBUG
		fprintf(dbgFile, "PipeServer: Waiting for Agent Named Pipe %s\n", handle->agentUUID.data);
		fflush(dbgFile);
		#endif
		while(!(ra_connectToNamedPipe(&handle->pipefd, handle->agentUUID.data)) && !handle->finalized) {
#if defined(MVS)
			fprintf(stdout, "Agent named pipe is not active: will wait in a %d ms loop\n", pipe_timeout);
			fflush(stdout);
#else
			printf("Agent named pipe is not active: will wait in a %d ms loop\n", pipe_timeout);
#endif
#ifdef _DEBUG
			fprintf(dbgFile, "Agent Named Pipe is not active: will wait in a %d ms loop\n", pipe_timeout);
			fflush(dbgFile);
#endif
			SLEEP(pipe_timeout);
		}
		#ifdef _DEBUG
		fprintf(dbgFile, "PipeServer: Agent Named Pipe %s has connected, fd = %d\n", handle->agentUUID.data, handle->pipefd);
		fflush(dbgFile);
		#endif

		/* Read our configuration info from the RAC. */
		messageLoop(handle, pipe_timeout); /* APAR PK01709: Configurable named pipe timeout */

		ra_disconnectFromNamedPipe(handle->pipefd); /* Bug 159179 */
		#ifdef _DEBUG
		fprintf(dbgFile, "PipeServer: disconnecting and closing named pipe %d\n", handle->pipefd);
		fflush(dbgFile);
		#endif

		#ifdef _DEBUG
		fprintf(dbgFile, "PipeServer: messageLoop has been broken due to RAC connection lost\n");
		fflush(dbgFile);
		#endif
	}

	/* Perform the thread cleanup */
#ifndef _WIN32
	pthread_cleanup_pop(1);
#endif

#ifdef _DEBUG
	fclose(dbgFile);
#endif

#ifdef _WIN32
	// Bug 278655 - The pipe server thread is about to terminate, so signal this in the agent handle
	handle->pipeServerThreadEnded = 1;
#endif
	return NULL;
}


#ifdef _WIN32
DWORD WINAPI win32pipeServerProxy(LPVOID args) {
	DWORD returnVal=0;
	PipeServer(args);
	return returnVal;
}
#endif


static void messageLoop(RA_AGENT_HANDLE agentHandle, int timeout) { /* APAR PK01709: Configurable named pipe timeout */
	int result;
	unsigned int bytesRead;
	unsigned char request[MESSAGE_BUFFER_INCREMENT];
	ra_message_t *message;
	unsigned char *correctBuffer;
	ra_uint_t messageLength, offset;
	BOOL sleepPrinted = FALSE; /* APAR PK01709: Configurable named pipe timeout */
	ra_command_t connectedCommand;

	/* Initialize our variables */
	message=NULL;
	offset=0;

	while(!agentHandle->finalized) {
		#ifdef _DEBUG
		fprintf(dbgFile, "messageLoop: About to read from Agent Named Pipe %s, fd = %d\n", agentHandle->agentUUID.data, agentHandle->pipefd);
		fflush(dbgFile);
		#endif

		result = readConnection(agentHandle->pipefd, request, offset, MESSAGE_BUFFER_INCREMENT, &bytesRead);

		// Check the connection is valid BEFORE we call the agentHandle->commandHandler function 
		// (when the JVM has already been shutdown, and the handler uses JNI code, it has a tendency to get stuck!)
		//
		// On Windows, the negative or zero result is, at program termination,
		// the result of the named pipe being closed by raStopListener.
		if(result <= 0) {
			/* Close the message pump */
			return; /* Let PipeServer to check for RAC connection */
		}
		/* Inform the agent we are connected */
		agentHandle->registered = TRUE;
		connectedCommand.tag = RA_AGENT_CONTROLER_AVAILABLE;
		ra_copyRASTRING(&(connectedCommand.info.agentName), &agentHandle->agentName);
		(agentHandle->commandHandler)(&connectedCommand);
		ra_destroyRASTRING((ra_string_t*)&connectedCommand.info.agentName);

		/* The connection is going down.  We need to break the loop */
		if(result <= 0) {
			/* Close the message pump */
			#ifdef _DEBUG
			fprintf(dbgFile, "messageLoop: readConnection on fd = %d failed, recheck RAC status, errno = %d\n", agentHandle->pipefd, ra_getLastErrorMinor());
			fflush(dbgFile);
			#endif
			return; /* Let PipeServer to check for RAC connection */
		}

		/* Increment bytesRead in case the master buffer contained a message fragment - 223065 */
		bytesRead+=offset;

		#ifdef _DEBUG
		fprintf(dbgFile, "messageLoop: fd = %d, bytesRead=%d  offset=%d\n", agentHandle->pipefd, bytesRead, offset);
		fflush(dbgFile);
		#endif

	    /* We assume the master buffer will not be fragmented due to an error later */
		offset=0;

	    /* Validate the header of this message.  If it is invalid go into
		   error recovery mode.  If it is valid then find out how long this
		   message is.
		*/
	    correctBuffer=request;
validate:
	    if(validateRecvBuffer(correctBuffer, bytesRead)) {
			BOOL agentControllerStillActive;
			messageLength=getMessageLengthfromValidBuffer(correctBuffer);

			#ifdef _DEBUG
			fprintf(dbgFile, "messageLoop: fd = %d, messageLength=%d  bytesRead=%d\n", agentHandle->pipefd, messageLength, bytesRead);
			fflush(dbgFile);
			#endif

			/* Are we missing part of the message because of overflow in the buffer */
			if(messageLength>bytesRead) {
				unsigned char *temp=(unsigned char*)ra_malloc(messageLength);
				unsigned int currentLength=bytesRead;

				/* Save the data we have so far. */
				memcpy(temp, correctBuffer, currentLength);
				
				while(currentLength < messageLength) {
					result = readConnection(agentHandle->pipefd, temp, currentLength, messageLength, &bytesRead);

					/* If we have lost a connection before a message is complete we should abandon */
					if(result <= 0) {
						#ifdef _DEBUG
						fprintf(dbgFile, "messageLoop: readConnection on fd = %d failed, recheck RAC status, errno = %d\n", agentHandle->pipefd, ra_getLastErrorMinor());
						fflush(dbgFile);
						#endif
						return; /* Let PipeServer to check for RAC connection */
					}
					currentLength += bytesRead;

					#ifdef _DEBUG
					fprintf(dbgFile, "messageLoop: fd = %d, currentLength = %d\n", agentHandle->pipefd, currentLength);
					fflush(dbgFile);
					#endif
				}

				message=ra_readMessageFromBuffer(temp, (unsigned long)messageLength);

				#ifdef _DEBUG
				fprintf(dbgFile, "messageLoop: fd = %d, after reading message from buffer 1: length=%d\n", agentHandle->pipefd, message->length);
				fflush(dbgFile);
				#endif

				/* Forward to the handler  */
				agentControllerStillActive=processControlMessage(agentHandle, message); 
				
				#ifdef _DEBUG
				fprintf(dbgFile, "messageLoop: fd = %d, after processing control message\n", agentHandle->pipefd);
				fflush(dbgFile);
				#endif

				ra_destroyMessage(message, TRUE);
				ra_free(temp);

				if(!agentControllerStillActive) {
					#ifdef _DEBUG
					fprintf(dbgFile, "messageLoop: fd = %d, Agent Controller is not active, exiting message loop\n", agentHandle->pipefd);
					fflush(dbgFile);
					#endif
					return;
				}
			}
			else {
				/* We have the entire message, but do we have more then one message */
				message=ra_readMessageFromBuffer(correctBuffer, (unsigned long)(bytesRead));

				#ifdef _DEBUG
				fprintf(dbgFile, "messageLoop: fd = %d, after reading message from buffer 2: length=%d\n", agentHandle->pipefd, message->length);
				fflush(dbgFile);
				#endif

				/* Forward to the handler */
				agentControllerStillActive=processControlMessage(agentHandle, message);  

				bytesRead=bytesRead-message->length;
				
				#ifdef _DEBUG
				fprintf(dbgFile, "messageLoop: fd = %d, after processing control message - bytesRead = %d\n", agentHandle->pipefd, bytesRead);
				fflush(dbgFile);
				#endif

				memcpy(correctBuffer, correctBuffer+message->length, bytesRead);
				ra_destroyMessage(message, TRUE);

				if(!agentControllerStillActive) {
					#ifdef _DEBUG
					fprintf(dbgFile, "messageLoop: fd = %d, Agent Controller is not active, exiting message loop\n", agentHandle->pipefd);
					fflush(dbgFile);
					#endif
					return; /* Let PipeServer to check for RAC connection */
				}

				#ifdef _DEBUG
				fprintf(dbgFile, "messageLoop: fd = %d, go back to validate remainder of message data\n", agentHandle->pipefd);
				fflush(dbgFile);
				#endif

				goto validate;
			}
		}
		else {
			#ifdef _DEBUG
			fprintf(dbgFile, "messageLoop: fd = %d, invalid buffer processing.\n", agentHandle->pipefd);
			fflush(dbgFile);
			#endif

			bytesRead=errorRecoveryScan(correctBuffer, bytesRead);

			#ifdef _DEBUG
			fprintf(dbgFile, "messageLoop: fd = %d, bytesRead after errorRecoveryScan = %d\n", agentHandle->pipefd, bytesRead);
			fflush(dbgFile);
			#endif

			/* If we have enough left that constitutes a header continue processing */
			if(bytesRead>=24) {
				goto validate;
			}
			else {
				/* We need to read some more data.  Keep what we have so far */
				if(correctBuffer!=request) {
					memcpy(request, correctBuffer, bytesRead);
				}
				offset=bytesRead;
			}
			#ifdef _DEBUG
			fprintf(dbgFile, "messageLoop: fd = %d, go back to read some more data\n", agentHandle->pipefd);
			fflush(dbgFile);
			#endif
		}
	}

#ifdef _DEBUG
	fprintf(dbgFile, "messageLoop: fd = %d, connection with RAC has been broken\n", agentHandle->pipefd);
	fflush(dbgFile);
#endif

	return;
}

/** READ_CONNECTIION  **********************************************************
  * Generic read function for all platforms for both sockets and pipes.
  * @returns  0 - connection closed gracefully
  */
static int readConnection(RA_HANDLE connection,
						  unsigned char *buffer,
						  int offset,
						  int length,
						  unsigned int *bytesRead) {
#if defined _HPUX || defined __OS400__
	return ra_readFromNamedPipe(connection, (char*)buffer, offset, length, (int*)bytesRead);
#else
	return ra_readFromNamedPipe(connection, buffer, offset, length, bytesRead);
#endif
}


/** 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;

}


/** ERROR_RECOVERY_SCAN  *****************************************************
  * When we get a bad message flow we go into recovery mode, searching for the
  * next occurance of the majic number in the stream.  The buffer is then
  * compressed and number of valid remaining bytes is returned.
  */
static ra_uint_t errorRecoveryScan(unsigned char *buffer, int length) {
	int offset;
	/* If there isn't enough bytes to check the majic number return zero */
	if(length<4) {
		return length;
	}
	/* Search for the next occurance of the majic number */
	for(offset=0; offset<length-3; offset++) {
		if(buffer[offset]==0x82 && buffer[offset+1]==0x65 && buffer[offset+2]==0x67 && buffer[offset+3]==0x80) {
			memcpy(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 */
	memcpy(buffer, buffer+offset, length-offset);
	return length-offset;

}


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





BOOL ra_initializeVMLock() {
#ifdef _WIN32
	_vmLock=CreateEvent(NULL,
					 TRUE,
					 TRUE,
					 NULL);
	if(_vmLock==NULL)
		return FALSE;
#else
	pthread_mutex_init(&_vmLock_mutex, NULL);
	pthread_cond_init(&_vmLock_cond, NULL);
#endif
	_pendingVMLock=FALSE;
	return TRUE;
}

void ra_setVMLock() {
#ifdef _WIN32
	ResetEvent(_vmLock);
#else
	pthread_mutex_lock(&_vmLock_mutex);
#endif
	_pendingVMLock=TRUE;
#ifndef _WIN32
	pthread_mutex_unlock(&_vmLock_mutex);
#endif
}

void ra_releaseVMLock() {
#ifndef _WIN32
	pthread_mutex_lock(&_vmLock_mutex);
#endif
	_pendingVMLock=FALSE;
#ifdef _WIN32
	SetEvent(_vmLock);
#else
	pthread_mutex_unlock(&_vmLock_mutex);
	pthread_cond_broadcast(&_vmLock_cond);
#endif
}

void ra_testVMLock() {
#ifndef _WIN32
	pthread_mutex_lock(&_vmLock_mutex);
#endif
	if(_pendingVMLock) {
#ifdef _WIN32
		WaitForSingleObject(_vmLock, INFINITE);
#else
		pthread_cond_wait(&_vmLock_cond, &_vmLock_mutex);
#endif
	}
#ifndef _WIN32
	pthread_mutex_unlock(&_vmLock_mutex);
#endif

}

BOOL ra_getVMLockState() {
	if(_pendingVMLock)
		return TRUE;
	else
		return FALSE;
}

static void addHandle(RA_AGENT_HANDLE handle) {
	agentHandleList[agentHandleCount++] = handle;
}

static void removeAllHandles() {
	int i;
	for(i = 0; i < agentHandleCount; i++) {
		if(agentHandleList[i] != NULL) {
			#ifdef _DEBUG
			if (dbgFile != NULL) {
				fprintf(dbgFile, "removeAllHandles: about to stop listener thread\n");
				fflush(dbgFile);
			}
			#endif
			ra_stopListener(agentHandleList[i]);
			/* Getting a SIGSEV on Red Hat when exiting workbench, disable the finalize for now. */
/*			ra_finalizeBindings(agentHandleList[i]);*/
		}
	}
}


static void removeHandle(RA_AGENT_HANDLE handle) {
	int i;
	for(i = 0; i < agentHandleCount; i++) {
		if(agentHandleList[i] == handle) {
			agentHandleList[i]=NULL;
			return;
		}
	}
}


static void sendAgentInactive(RA_AGENT_HANDLE handle) {
	ra_message_t *message;
	ra_command_t *command;

	message=ra_createMessage(RA_CONTROL_MESSAGE, 0);
	command=ra_addCommandToMessage(message, NULL);


	/* Set the agent information */
	command->tag=RA_AGENT_INACTIVE;
	command->info.agent_inactive.processId = ra_getProcessId();
	ra_createRASTRING(&command->info.agent_inactive.processUUID, ra_getProcessUUID(handle));
	ra_copyRASTRING(&command->info.agent_inactive.agent, &handle->agentName);
	ra_createRASTRING(&command->info.agent_inactive.agentUUID, ra_getAgentUUID(handle));
	ra_copyRASTRING(&command->info.agent_inactive.agentType, &handle->agentType);

	/* Inform server we are exiting */
	ra_sendMessage(handle, message);

	/* Clean up the allocated memory */
	ra_destroyMessage(message, TRUE);

}

void ra_getPropertyValues(RA_AGENT_HANDLE handle, char* name, char* type) {
	ra_message_t *message;
	ra_command_t *command;

	message = ra_createMessage(RA_CONTROL_MESSAGE, 0);
	command = ra_addCommandToMessage(message, NULL);

	command->tag = RA_GET_PROPERTY_LIST;
	command->info.query_property_list.context = 0;
	ra_createRASTRING(&command->info.query_property_list.name, name);
	ra_createRASTRING(&command->info.query_property_list.type, type);
	ra_createRASTRING(&command->info.query_property_list.agentUUID, ra_getAgentUUID(handle));

	ra_sendMessage(handle, message);

	ra_destroyMessage(message, TRUE);

	return;
}
