/*******************************************************************************
 * 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:
 *    Hoang M Nguyen, Intel - Initial API and Implementation
 *
 * $Id: NamedPipeListener.c,v 1.57 2009/09/24 21:02:06 jwest Exp $
 *
 *******************************************************************************/ 


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

Each transport layer will provide and implement exactly the same 
pre-defined interface. Each will be in its own DLL or shared library
that is dynamically loaded at runtime.

The interaction between Agent Controller (AC) and the transport layer (TL)
is defined in "MsgPipeline.h" header file.

- from AC to TL, functions include "createTransportListener(), 
  destroyTransportListener(), etc.

- from TL to AC, see function callback in "transport_layer_data_t" structure.

  ------
  |    |
  | AC |   
  |    |<-------|
  ------        |
                V
              ------
              |    |
              | TL |
              |    |
              ------
               ^  ^
         |-----|  |---------------|
         |                        |
         V                        V
    ----------                ---------
    |        |                |       |
    | Client |                | Agent |
    |        |                |       |
    ----------                ---------


The interaction between the transport layer (TL) and the client
or the agent is defined in "TPTPMessageHeader.h" (CONNECT, DISCONNECT, etc.)
and basically any command or data.

The TL will process only CONNECT and DISCONNECT commands and forward
everything else to the AC for routing to the right destination for processing.

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


#include "NamedPipeListener.h"
#include "tptp/TransportSupport.h"
#include "tptp/TPTPSupportUtils.h"
#include "tptp/TPTPUtils.h"
#include "NamedPipeTLLog.h"
#include "tptp/dime.h"


/** maximum number of retries to connect to the master named pipe */
#define MAX_RETRIES 64 


/** thread status */
enum ThreadStatus { IDLE, RUNNING } ;

/*====================================================================*/
/*    internal function prototypes                                    */
/*====================================================================*/
int    waitForConnection(HANDLE *pPipeHandle, server_block_ptr_t pServerData); 

request_block_ptr_t  getInitRequestDataBlock(server_block_ptr_t pServerData, int connType) ;

THREAD_USER_FUNC_RET_TYPE doListening(LPVOID args) ;

int processCONNECTCall(server_block_t* pServerData, char *pUuid, int uuidLength);

int processLAUNCHEDCall(server_block_t* pServerData, tptp_console_launch_cmd_t *pLaunchedCmd) ;

THREAD_USER_FUNC_RET_TYPE processConsoleRequest(LPVOID args) ;

int prepareForConsole(request_block_ptr_t  pBlk) ;

int forwardDataToPartner(request_block_ptr_t pBlk, int dataLen, char *pBuffer) ;

int forwardDataWithDIMEToPartner(request_block_ptr_t pBlk, int dataLen, char *pBuffer, int consoleType) ;

//int processRecdMsgs(server_block_ptr_t pServerData, char *pBuffer, int bufferLength, int bytesRead) ;
int processRecdMsgs( server_block_ptr_t pServerData, 
                     char*            pBuffer, 
                     unsigned int     bufferLength, 
					 unsigned int     bytesReadIn, 
					 unsigned int*    offset, 
					 unsigned int*    messageLenOut);

int processOneMsg(server_block_ptr_t pServerData, char *pCmd) ;


/**
 *********************************************************
 *
 * @brief
 *    create and initialize the data block used to process this new request
 *
 * @return
 *    Pointer to the data block used to process this client request
 *    NULL if error
 *********************************************************/
request_block_ptr_t  getInitRequestDataBlock(server_block_ptr_t pServerData, int connType)
{
	int connId = 0 ;

	HashTable           *pTab = NULL ;

	request_block_ptr_t pRequestDataBlock = NULL ;

	/* set up the data block for each request */
	pRequestDataBlock = (request_block_ptr_t) tptp_malloc(sizeof(request_block_t)) ;
	BZERO( pRequestDataBlock, sizeof(request_block_t) );
	pRequestDataBlock->pServerData = pServerData ;
	pRequestDataBlock->connectionId = 0 ;

	pRequestDataBlock->connectionType = 0 ;
	pRequestDataBlock->processID4Console = TPTP_INVALID_PID ;

	pRequestDataBlock->consoleInputPipe = 0 ;


	/* tell the agent controller about the new connection  */
	/*    and receive the assigned connection id           */
	if (connType == CONTROL_CHANNEL)
	{
		addConnectionEntry_ptr_t pFunc = NULL ;
		pFunc = pServerData->agentControllerDataBlk.addConnectionEntry ;
		pFunc(pServerData->cmo, pServerData->agentControllerDataBlk.transportID, &(pRequestDataBlock->connectionId)) ;
	}
	else if ((connType == DATA_CHANNEL) || (connType == CONSOLE_CHANNEL))
	{
		addDataConnectionEntry_ptr_t pDataFunc = NULL ;
		unsigned int direction = SEND_DATA | RECEIVE_DATA ;

		pDataFunc = pServerData->agentControllerDataBlk.addDataConnectionEntry ;
		pDataFunc(pServerData->cmo, pServerData->agentControllerDataBlk.transportID,
			direction,
			&(pRequestDataBlock->connectionId)) ;
		TPTP_LOG_DEBUG_MSG(pServerData, "Adding new DATA or CONSOLE channel.") ;
	}

	/* save the control block away */
	/*   use the connection id as the index into the array for fast lookup */
	connId = pRequestDataBlock->connectionId ;
	TPTP_LOG_DEBUG_MSG1(pServerData, "NamedPipe getInitRequestDataBlock: returning Connection ID: %d.", connId) ;

	/* add the request in the connection table */
	pTab = pServerData->connectionTable ;
	tablePut(pTab, connId, (Entry_value_ptr_t) pRequestDataBlock) ;

	return ( pRequestDataBlock ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    send a given message to the given destination (identified by connection id)
 *
 * @return
 *    Negative - Error
 *    0 - Socket is closed gracefully
 *    Positive - the number of bytes that have been sent
 *********************************************************/

int   sendThisMessage( server_block_t* pServerData, int connectionID, int cmdSize, char * pCmdBlock, BOOL shouldAddHeader)
{
	int rc = 0 ;
	HANDLE *pAgentPipe = NULL ;
	int  bytesSent  = 0 ;
	char *buffer = NULL ;
	int  bufferLength = 0 ;
	unsigned int flags = 0 ;
	request_block_ptr_t pBlock;
	char *pSendBuffer = NULL ;

	/* locate the sender's block */
	pBlock = (request_block_ptr_t) tableGet(pServerData->connectionTable, connectionID) ;

	/* locate the socket to send */
	pAgentPipe = &pBlock->agentPipe ;

	if (shouldAddHeader == TRUE)
	{
		/* add the header to the message if requested */
		addBasicMsgHeader(pCmdBlock, cmdSize, &buffer, &bufferLength, flags);
		pSendBuffer = buffer;
	}
	else
	{
		pSendBuffer = pCmdBlock;
		bufferLength = cmdSize ;
	}

	/* sent a message to the agent controller (server) */
	rc = writeToNamedPipe(*pAgentPipe, pSendBuffer, 0, bufferLength, &bytesSent);
	if (bytesSent < 0)
	{
		TPTP_LOG_ERROR_MSG1(pServerData, "NamedPipe: Failed to send data on connection ID %d", connectionID);
	}
	else
	{
		TPTP_LOG_DEBUG_MSG2(pServerData, "NamedPipe: Sent %d bytes on connection ID %d", bytesSent, connectionID);
		printThisEnvelope((tptp_basic_msg_header_ptr_t) pSendBuffer) ;
		rc = bytesSent ;
	}

	if (buffer) tptp_free(buffer);

	return ( rc ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    prepare console I/O support for an application:
 *    - open the stdin named pipe of the application for
 *      writing data into. The input data will come across the
 *      console data connection
 *    - start 2 threads for handling data coming through
 *      the application's stdout and stderr pipes.    
 *
 * @return
 *    Always 0
 *********************************************************/

int prepareForConsole(request_block_ptr_t  pBlk)
{
	int       rc = 0 ;
	TID threadId;
	HANDLE threadHandle ;
	char *pUuidPart = NULL ;
	char uuid[TPTP_DEFAULT_UUID_MAX_LENGTH] ;

	console_thread_block_ptr_t pThreadData = NULL ;

	/* start thread to handle console stdout */
	pThreadData = (console_thread_block_ptr_t) malloc(sizeof(console_thread_block_t)) ;
	pThreadData->pRequestBlk = pBlk ;
	pThreadData->threadStatus = TPTP_IDLE_THREAD ;
	pThreadData->consoleType = TPTP_STDOUT ;

	rc = tptpStartThread(processConsoleRequest,
		(LPVOID) pThreadData, &threadId, &threadHandle) ;
	CLOSE_THREAD_HANDLE(threadHandle);


	/* start thread to handle console stderr */
	pThreadData = (console_thread_block_ptr_t) malloc(sizeof(console_thread_block_t)) ;
	pThreadData->pRequestBlk = pBlk ;
	pThreadData->threadStatus = TPTP_IDLE_THREAD ;
	pThreadData->consoleType = TPTP_STDERR ;

	rc = tptpStartThread(processConsoleRequest,
		(LPVOID) pThreadData, &threadId, &threadHandle) ;
	CLOSE_THREAD_HANDLE(threadHandle);


	/* Open the write end of the console stdin named pipe.
	 * Input to the application which comes across the console
	 * data channel will be written into this pipe.
	 */
	sprintf(uuid, "%s%s", pBlk->fullPipeName, "-stdin") ;
	pUuidPart = uuid + strlen(RA_PIPE_NAMESPACE) ;
	pBlk->consoleInputPipe = openWriteOnlyNamedPipe(RA_PIPE_NAMESPACE, pUuidPart, TRUE) ;

	return 0 ;
}



/**
 *********************************************************
 *
 * @brief
 *    receive and process incoming request message
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
THREAD_USER_FUNC_RET_TYPE processConsoleRequest(LPVOID args) 
{
	int    rc = 1;

	HANDLE   consoleHandle = 0 ;

	char uuid[TPTP_DEFAULT_UUID_MAX_LENGTH] ;
	char *pUuidPart = NULL ;

	unsigned int  bytesRead;

	unsigned char buffer[TPTP_DEFAULT_BUFFER_MAX_LENGTH];
	unsigned int  length = TPTP_DEFAULT_BUFFER_LENGTH ;

	console_thread_block_ptr_t pThreadData = (console_thread_block_ptr_t) args ;

	/* set up environmental info for this incoming message */
	request_block_ptr_t pRequestDataBlock = pThreadData->pRequestBlk ;

	pUuidPart = uuid + strlen(RA_PIPE_NAMESPACE) ;

	switch(pThreadData->consoleType)
	{
		case TPTP_STDOUT: 
			sprintf(uuid, "%s%s", pRequestDataBlock->fullPipeName, "-stdout") ;
			consoleHandle = openReadOnlyNamedPipe(RA_PIPE_NAMESPACE, pUuidPart, TRUE) ;
			break ;
		case TPTP_STDERR: 
			sprintf(uuid, "%s%s", pRequestDataBlock->fullPipeName, "-stderr") ;
			consoleHandle = openReadOnlyNamedPipe(RA_PIPE_NAMESPACE, pUuidPart, TRUE) ;
			break ;
		default: ;
	}

	/* initial status before the thread is running */
	pThreadData->threadStatus = TPTP_RUNNING_THREAD ;

	TPTP_LOG_DEBUG_MSG(pThreadData->pRequestBlk->pServerData, "start console thread (named pipe)...") ;

	/* accept and process one connection at a time */
	while ((pThreadData->threadStatus == TPTP_RUNNING_THREAD) && (rc > 0))
	{
		switch(pThreadData->consoleType)
		{
			case TPTP_STDOUT: 
			case TPTP_STDERR:
				/* read the message */
				TPTP_LOG_DEBUG_MSG(pThreadData->pRequestBlk->pServerData, "Console: waiting for data from stdout.") ;
				TPTP_LOG_DEBUG_MSG1(pThreadData->pRequestBlk->pServerData, "About to read named pipe. Handle(%x).", consoleHandle) ;

				rc = readFromNamedPipe(consoleHandle, buffer, 0, length, &bytesRead);
				// TPTP_LOG_DEBUG_MSG2(pThreadData->pRequestBlk->pServerData, "readFromNamedPipe: rc - %d, data - %s, bytesRead - %d\n", rc, buffer, bytesRead);

				if (rc >= 0)
				{
					buffer[bytesRead] = '\0' ;
					TPTP_LOG_DEBUG_MSG2(pThreadData->pRequestBlk->pServerData, "Console: data(%s) length(%d).", buffer, bytesRead) ;
				}
				else if (rc == TPTP_PIPE_HAS_ENDED)
				{
					/* this is expected and normal */
					TPTP_LOG_DEBUG_MSG(pThreadData->pRequestBlk->pServerData, "Console: End of pipe recieved - exiting");
					pThreadData->threadStatus = TPTP_IDLE_THREAD;
				}
				else
				{
					TPTP_LOG_DEBUG_MSG2(pThreadData->pRequestBlk->pServerData, "Console: Error reading pipe(%s) rc(%d).", uuid, rc) ;
					pThreadData->threadStatus = TPTP_IDLE_THREAD;
				}

				if (rc > 0)
				{
					if (pRequestDataBlock->processID4Console == TPTP_INVALID_PID)
					{
						tptp_waitSemaphore(&pRequestDataBlock->hasProcessIdSemaphore) ;
					}
				}

				
				/* send the stdout/stderr console output through the data connection */
				if (rc > 0)
				{
					TPTP_LOG_DEBUG_MSG3(pThreadData->pRequestBlk->pServerData, "CONSOLE THREAD: sending console output data(%s) length(%d) for pid(%lu)", buffer, bytesRead, (unsigned long)pRequestDataBlock->processID4Console);
					forwardDataWithDIMEToPartner(pRequestDataBlock, bytesRead, buffer, pThreadData->consoleType) ;
				}

				break ;
		}
	}

	TPTP_LOG_DEBUG_MSG(pThreadData->pRequestBlk->pServerData, "Console: Closing named pipes") ;
    
	rc = disconnectFromNamedPipe(consoleHandle);
	TPTP_LOG_DEBUG_MSG1(pThreadData->pRequestBlk->pServerData, "Disconnect from named pipe %d", rc);
	cleanPipeUp(&consoleHandle);

	disconnectFromNamedPipe(pRequestDataBlock->consoleInputPipe);
	cleanPipeUp(&pRequestDataBlock->consoleInputPipe);

	#if !defined(MVS)
		tptp_deleteSemaphore(&pRequestDataBlock->hasProcessIdSemaphore);
	#else
		// On MVS, this semaphore is deleted on process termination.
	#endif
	
	

	TPTP_LOG_DEBUG_MSG(pThreadData->pRequestBlk->pServerData, "Stdout/stderr listening thread exited");

	free(pThreadData) ;

	return ( 0 ) ;
}

/**
 *********************************************************
 *
 * @brief
 *    Process the CONNECT request
 *
 * @return
 *    0 - success
 *    non-zero - error occurs
 *********************************************************/
int processCONNECTCall(server_block_t* pServerData,  
					   char *pUuid, int uuidLength)
{
	int rc = 0 ;
	char  *buffer = NULL;
	int   bufferLength =0;
	BOOL  shouldAddHeader = FALSE ;

	unsigned int flags = 0 ;

	char *cmd = NULL;
	int   cmdLength = 0;
	char uuid[TPTP_DEFAULT_UUID_MAX_LENGTH] ;

	/* set up the data block for the new request */
	request_block_ptr_t pBlk = getInitRequestDataBlock(pServerData, CONTROL_CHANNEL) ;


	/* keep the agent's pipe open for sending data */
	strncpy(uuid, pUuid, uuidLength) ;
	uuid[uuidLength] = '\0' ;
	TPTP_LOG_DEBUG_MSG1(pServerData, "UUID: %s.", uuid) ;

	pBlk->agentPipe = openWriteOnlyNamedPipe(RA_PIPE_NAMESPACE, uuid, FALSE) ;

	/* Save this agent's pipe name */
	strcpy(pBlk->fullPipeName, RA_PIPE_NAMESPACE);
	strcat(pBlk->fullPipeName, uuid);

	/* return the assigned connection id */
	cmdLength = sizeof(int);
	cmd = (char*)tptp_malloc(cmdLength);
	if (!cmd) return -1;
	writeUINTToBuffer(cmd, pBlk->connectionId);

	/* prepare the response to the CONNECT call */
	flags = CONNECTION_COMPLETE ;
	addBasicMsgHeader(cmd, cmdLength, &buffer, &bufferLength, flags) ;
	
	/* send the response */
	sendThisMessage(pServerData, pBlk->connectionId, bufferLength, buffer, shouldAddHeader) ;

	if (cmd) tptp_free(cmd);
	if (buffer) tptp_free(buffer);
	return ( rc ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    Process the CONSOLE_CONNECT request
 *
 * @return
 *    0 - success
 *    non-zero - error occurs
 *********************************************************/
int processCONSOLE_CONNECTCall(server_block_t* pServerData, tptp_console_connect_cmd_t * pConsoleConnCmd)
{
	int rc = 0;
	char  *buffer = NULL;
	int   bufferLength = 0;
	BOOL  shouldAddHeader = FALSE;

	unsigned int flags = 0 ;

	char *cmd = NULL;
	int   cmdLength = 0;

	/* set up a data block which results in a connectionId being assigned to it*/
	request_block_ptr_t pBlk = getInitRequestDataBlock(pServerData, CONSOLE_CHANNEL) ;

	/* Save build the common portion of the string to use in creating the
	 * names of the console pipes.  The -stdin, -stdout, -stderr suffix will be
	 * be added to this unique string to form the full pipe name which was
	 * opened by the sender of the connect request.
	*/
	sprintf(pBlk->fullPipeName, "%s%s", RA_PIPE_NAMESPACE, pConsoleConnCmd->uuid) ;

	TPTP_LOG_DEBUG_MSG1(pServerData, "Console UUID: %s.", pConsoleConnCmd->uuid) ;

	/* console uses data channel (no interpretation) */
	pBlk->connectionType = DATA_CHANNEL ;


	/* Save the connection Id to use in sending the response cmd */
	pBlk->replyConnId   = pConsoleConnCmd->replyConnId ;

    tptp_initializeSemaphore(& pBlk->hasProcessIdSemaphore);
    
    #ifdef MVS
    	// hasProcessIdSemaphore is not deleted properly, so we register it as one that must be deleted on process shutdown
    	tptp_zosRegisterUndeletedSemaphore(& pBlk->hasProcessIdSemaphore);
	#endif

	TPTP_LOG_DEBUG_MSG1(pServerData, "CONSOLE CONNECT:established new connectionId %d", pBlk->connectionId) ;

	/* return the connection Id in the completion response message */
	cmdLength = sizeof(int);
	cmd = (char*) tptp_malloc(cmdLength);
	if (!cmd) return -1;

	writeUINTToBuffer(cmd, pBlk->connectionId);

	flags = CONSOLE_CONNECT_COMPLETE ;
	addBasicMsgHeader(cmd, cmdLength, &buffer, &bufferLength, flags);
	
	/* send the response */
	sendThisMessage(pServerData, pBlk->replyConnId, bufferLength, buffer, shouldAddHeader) ;

/* TODO: Logic here is faulty.  The named pipes should be created before the
 * CONSOLE_CONNECT_COMPLETE is sent.  If there was a failure, we should instead
 * be sending s CONSOLE_CONNECT_FAILED.  The call to prepareForConsole() should have
 * an error return check.
*/
	/* Create Named pipes for console */
	prepareForConsole(pBlk) ;

	if (cmd) tptp_free(cmd);
	if (buffer) tptp_free(buffer);

	return ( rc ) ;
}



/**
 *********************************************************
 *
 * @brief
 *    handle the DISCONNECT request
 *
 *********************************************************/
int handleDISCONNECT(server_block_ptr_t pServerData, char *pMsg)
{
	tptp_uint32 connectionID ;

	/* read in the given conn id */
	pMsg = readUINTFromBuffer(pMsg, &connectionID);

	terminateNamedPipeConnection(pServerData, connectionID) ;

	return 0 ;
}


/**
 *********************************************************
 *
 * @brief
 *    handle the PROCESS LAUNCHED command for console support
 *
 *********************************************************/

int processLAUNCHEDCall(server_block_t* pServerData, tptp_console_launch_cmd_t *pLaunchedCmd)
{
	unsigned int processConnId = pLaunchedCmd->processConnId ;

	/* locate the sender's block */
	request_block_ptr_t pBlock = 
		(request_block_ptr_t) tableGet(pServerData->connectionTable, processConnId) ;

	TPTP_LOG_ERROR_MSG1(pServerData, "CONSOLE LAUNCHED COMMAND for connection(%d). ", processConnId) ;

	if (pBlock == NULL)
	{
		TPTP_LOG_ERROR_MSG1(pServerData, "CONSOLE LAUNCHED failed for connection(%d). ", processConnId) ;
	}
	else
	{
		pBlock->processID4Console = pLaunchedCmd->processId ;
		TPTP_LOG_DEBUG_MSG1(pServerData, "Console received processid(%lu). ", (unsigned long)pBlock->processID4Console) ;

		tptp_postSemaphore(&pBlock->hasProcessIdSemaphore) ;
	}

	return 0 ;
}


/**
 *********************************************************
 *
 * @brief
 *    Extract out the message header from the given message
 *
 * @return
 *    The pointer to the message without the header
 *
 *    NULL if the message is intended to the transport layer
 *    and it has been processed and responded by the transport.
 *
 *********************************************************/
char * processTheMessageHeader(server_block_ptr_t pServerData, char *pMsg, int *pPayLoadLength)
{
	unsigned int magicNumber = 0;
	unsigned int flags = 0 ;

	/* read in the magic number */
	pMsg = readUINTFromBuffer(pMsg, &magicNumber);

	/* read in the flags */
	pMsg = readUINTFromBuffer(pMsg, &flags);

	/* read in the payload length */
	pMsg = readUINTFromBuffer(pMsg, pPayLoadLength) ;

	//printf("Processing the message header with payload length = %d and flags=0x%08x\n", *pPayLoadLength, flags);

	if ((flags & CONNECT) != 0)
	{
		/* CONNECT command. Go handle it. */
		//printf("This is the connect call\n");
		processCONNECTCall(pServerData, pMsg, *pPayLoadLength) ;

		/* prevent it from forwarding to the AC */
		pMsg = NULL ;
	}
	else if ((flags & CONNECT_CONSOLE) != 0)
	{
		tptp_console_connect_cmd_t * pConsoleConnCmd = NULL ;

		pConsoleConnCmd = (tptp_console_connect_cmd_t *) pMsg ;

		/* CONNECT command. Go handle it. */
		processCONSOLE_CONNECTCall(pServerData, pConsoleConnCmd) ;

		/* prevent it from forwarding to the AC */
		pMsg = NULL ;
	}
	else if ((flags & CONSOLE_PROCESS_LAUNCHED) != 0)
	{
		tptp_console_launch_cmd_t *pLaunchedCmd = (tptp_console_launch_cmd_t *) pMsg ;

		/* Launched command. Go handle it. */
		processLAUNCHEDCall(pServerData, pLaunchedCmd) ;

		/* prevent it from forwarding to the AC */
		pMsg = NULL ;

	}
	else if ((flags & DISCONNECT) != 0)
	{
		handleDISCONNECT(pServerData, pMsg) ;

		/* prevent it from forwarding to the AC */
		pMsg = NULL ;
	}


	return ( pMsg ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    process a given complete message 
 *
 *********************************************************/
int processOneMsg(server_block_ptr_t pServerData, char *pCmd)
{
	char * pMsg = NULL ;

	processMessage_ptr_t pFunc = NULL ;

	unsigned int payLoadLength = 0 ;

	/* Strip out the message header */
	pMsg = processTheMessageHeader(pServerData, pCmd, &payLoadLength) ;

	//printf("The Message received is - %s and length is %d\n", pMsg, payLoadLength);

	if (pMsg != NULL)
	{
		pMsg[payLoadLength] = '\0' ;

		//printf("Calling process message in processOneMsg\n");
		pFunc = pServerData->processMessage ;
		pFunc(pServerData->nexto, payLoadLength, pMsg) ;
	}

	return 0 ;
}

/**
 *********************************************************
 *
 * @brief
 *    process received messages by examining the buffer.
 *    This will process one message/command at a time
 *    until the end of the received buffer.
 *
 *    If there is incomplete command, it will move to the beginning
 *    of the buffer.
 *
 * @return
 *    0 - process all messages
 *    negative - Error.
 *    positive - number of bytes left to be processed
 *               These bytes have been copied out in front of the buffer
 *
 *********************************************************/
int processRecdMsgs( server_block_ptr_t pServerData, 
                     char*            pBuffer, 
                     unsigned int     bufferLength, 
					 unsigned int     bytesReadIn, 
					 unsigned int*    offset, 
					 unsigned int*    messageLenOut)
{

	char     *pMsg = NULL ;
	char* pBufferStart = pBuffer;
	unsigned int magicNumber = 0;
	unsigned int flags = 0 ;
	unsigned int payLoadLength = 0 ;
	unsigned int msgLength = 0 ;

	unsigned int bytesRead = (unsigned int)bytesReadIn;

	if ( ( offset == NULL ) || (messageLenOut == NULL) )
	{
		// TODO return proper error
		return -1;
	}

	//printf("reached here 1\n");

	while ( bytesRead > 0 )
	{		

		/* If we have enough left that constitutes a header continue processing */
		if ((bytesRead + *offset) < sizeof(tptp_basic_msg_header_t))
		{
			/* too small to determine the payload length. Go read some more */
			*offset += bytesRead ;			
			*messageLenOut = sizeof(tptp_basic_msg_header_t);
			return TPTP_NAMED_PIPE_PARTIAL_MESSAGE;
		}

		// read in the magic number 
		pMsg = readUINTFromBuffer(pBufferStart, &magicNumber);

		// read in the flags 
		pMsg = readUINTFromBuffer(pMsg, &flags);

		// read in the payload length 
		pMsg = readUINTFromBuffer(pMsg, &payLoadLength) ;

		msgLength = payLoadLength + sizeof(tptp_basic_msg_header_t) ;

		//printf("message received in processRecdMsgs() of lenth %d and payLoadLength %d and message %s and flags 0x%08x\n", bufferLength, payLoadLength, pMsg, flags);

		TPTP_LOG_DEBUG_MSG1( pServerData, "Validated message of %d bytes", msgLength );

		/* Are we missing part of the message because of overflow in the buffer */
		if ( msgLength > bytesRead)
		{
			/* Return this and the serveRequests thread will read more for us */
			*offset = bytesRead;
			*messageLenOut = msgLength;  
			return TPTP_NAMED_PIPE_PARTIAL_MESSAGE;
		}

		
		// we have at least one complete message, consume it 
		//printf("Calling the process message\n");
		processOneMsg(pServerData, pBuffer) ;


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

	*offset = 0;
	*messageLenOut = 0;

	return 0;
}



/**
 *********************************************************
 *
 * @brief
 *    main running thread that accepts connection,
 *    set up the environment and process it in the current thread
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
int serveRequest(server_block_ptr_t pServerData)
{
	int      rc = 0 ;
	int      bytesRead = 0 ;
	unsigned int      offset = 0 ;
	HANDLE   * pPipe       = NULL ;
	BOOL     isSuccessful  = FALSE ;

	unsigned char* buffer;
	unsigned int   bufferLength = TPTP_DEFAULT_BUFFER_LENGTH;

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

	/* initial status before the thread is running */
	pServerData->threadStatus = RUNNING ;
	pPipe                     = & pServerData->serverPipe ;

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

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

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

				rc = processRecdMsgs(pServerData, buffer, bufferLength, bytesRead+offset, &offset, &neededMsgLen);
				if ( rc == TPTP_NAMED_PIPE_PARTIAL_MESSAGE )
				{
					if ( neededMsgLen > bufferLength )
					{
						unsigned char* temp = buffer;
						
						TPTP_LOG_DEBUG_MSG1(pServerData, "Overfolow -- reallocating message buffer (%d bytes).", neededMsgLen+1 );

						buffer = tptp_realloc( buffer, neededMsgLen+1 );
						if ( buffer == NULL )
						{
							/* We can't allocate more memory, so we're going to lose this message
							 *    but we may be able to salvage the situation by emptying our
							 *    buffer.  The error scan in the process will find the start of the
							 *    next message.
							 */
							TPTP_LOG_ERROR_MSG(pServerData, "Unable to re-allocate message buffer.  The current message will be lost");
							buffer = temp;
							offset = 0;
						}
						else
						{
							/* 'offset' was set by processRecdMsgs */
							bufferLength = neededMsgLen;
						}
					}
				}
				else
				{
					offset = 0;
				}
			}

			disconnectFromNamedPipe(*pPipe);

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

		}
	}

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

	cleanPipeUp(pPipe);

	return ( rc ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    set up the server named pipe for listening incoming requests
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
THREAD_USER_FUNC_RET_TYPE doListening(LPVOID args) 
{
	server_block_ptr_t  pParam = (server_block_ptr_t) args ;

	TPTP_LOG_DEBUG_MSG1(pParam, "Named pipe server is running and listening at \"%s\" pipe.", pParam->fullPipeName) ;

	serveRequest(pParam) ;

	return ( 0 );
}

/**
 *********************************************************
 *
 * @brief
 *    create an instance of the named pipe listener
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
tptp_int32 createNamedPipeListener(tptp_object* cmo, transport_layer_data_t * pTransportData, tptp_object* tlo)
{
	int             rc = 0 ;
	char*           pipeName;
	server_block_t* pServerData = NULL ;

	/* We can't use the logging macros here because we haven't set up our server data structure yet */
	if ( pTransportData->logEventEntry )
	{
		pTransportData->logEventEntry( cmo, "Named Pipe TL", pTransportData->transportID, __FILE__, __LINE__, TPTP_DEBUG, "createTransportListener (named pipe)." );
	}

	/* prepare the globally available server data block */
	pServerData = (server_block_ptr_t) malloc(sizeof(server_block_t)) ;
	pServerData->cmo = cmo;
	pServerData->serverPipe = NULL ;
	pServerData->threadStatus = 0 ;
	pServerData->agentControllerDataBlk = *pTransportData ;

	/* Get the pipe name */
	rc = getNamedPipeConfigInfo(pTransportData->configurationData, &pipeName);
	if (rc == 0) 
	{
		strcpy(pServerData->fullPipeName, RA_PIPE_NAMESPACE);
		strcat(pServerData->fullPipeName, pipeName);
		tptp_free( pipeName );
	}
	else
	{
		strcpy(pServerData->fullPipeName, RA_PIPE_NAMESPACE);
		strcat(pServerData->fullPipeName, PIPE_DEFAULT_ADDRESS);
	}

	/* allocate connection table */
	pServerData->connectionTable = tableCreate();

	/* Save our state data */
	tlo->objectID = NAMED_PIPE_LISTENER_OBJECT_ID;
	tlo->data = pServerData;

	return ( rc ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    destroy an instance of the named pipe listener
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
tptp_int32 destroyNamedPipeListener(tptp_object* tlo)
{
	server_block_t* pServerData = tlo->data;

	/* Clear the object data */
	tlo->data = 0;
	tlo->objectID = -1;

	TPTP_LOG_DEBUG_MSG(pServerData, "destroyTransportListener (named pipe).") ;

	/* stop the running thread */
	pServerData->threadStatus = IDLE ;

	/* stop accepting connection */
	cleanPipeUp(&pServerData->serverPipe) ;

	/* TODO: Make sure that our threads have finished? */

	/* free the connection table */
    tableDelete( pServerData->connectionTable );
	pServerData->connectionTable = NULL ;

	/* this is to clean up the master pipe under Linux */
	destroyNamedPipe(pServerData->fullPipeName) ;

	/* free the global block */
	free( pServerData );

	return ( 0 ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    set the function for forwarding messages
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
tptp_int32 setNamedPipeProcessMessageFunc( server_block_t* pServerData, tptp_object* nexto, processMessage_ptr_t func)
{
	pServerData->processMessage = func;
	pServerData->nexto = nexto;

	return 0;
}


/**
 *********************************************************
 *
 * @brief
 *    create a new running thread to handle named pipe connections
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
int startNamedPipeListener(server_block_t* pServerData)
{
	int       rc;
	TID       threadId;
	HANDLE    threadHandle;
	char*     pBasePipeName = NULL;
	HANDLE    handle;

	/* TODO: determine whether or not the thread is already running and simply idle */
	
	/* Create the named pipe */
	pBasePipeName = pServerData->fullPipeName + strlen(RA_PIPE_NAMESPACE) ;
	handle = createReadOnlyNamedPipe(RA_PIPE_NAMESPACE, pBasePipeName, FALSE);
	
	if (handle == INVALID_HANDLE_VALUE)
	{
		TPTP_LOG_ERROR_MSG( pServerData, "Unable to create named pipe." );
		rc = -1;
	}
	else
	{
		setHandleInherited(handle);
		
		pServerData->serverPipe = handle;
	
		/* create new thread to listen to connection request */
		rc = tptpStartThread(doListening, (LPVOID)pServerData, &threadId, &threadHandle);
		CLOSE_THREAD_HANDLE(threadHandle);

		/* TODO: save the threadID and threadHandle? */
	}

	return rc;
}



/**
 *********************************************************
 *
 * @brief
 *    accept new request from this named pipe
 *
 * @return
 *    TRUE - Success
 *    FALSE - Error.
 *********************************************************/
int waitForConnection(HANDLE *pPipeHandle, server_block_ptr_t pServerData)
{
	int isConnectedSuccessful = FALSE ;

	int  retryCount=0;

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

	/* Wait for a connection. */
	/* We will only try and open the file up to 64 times.  
	 * If we cannot access our master pipe we will exit the server
	 */
	do 
	{
		isConnectedSuccessful = connectToNamedPipe(pPipeHandle, pServerData->fullPipeName);

		if (isConnectedSuccessful != 0)
			retryCount++;

	} while ((isConnectedSuccessful != 0) && (retryCount < MAX_RETRIES) && 
		(pServerData->threadStatus == RUNNING));

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

	return ( isConnectedSuccessful == 0 ) ;
}

int closeConnections(server_block_t* pServerData) {
	int i, tabSize;
	tptp_uint32* list;
	request_block_ptr_t pBlk;

	tabSize = tableSize(pServerData->connectionTable);
	if (tabSize <= 0) return 0;
	
	list = (tptp_uint32*) malloc(tabSize*sizeof(tptp_uint32));
	tableKeysAsList(pServerData->connectionTable, list, tabSize);
		
	for (i=0; i<tabSize; i++) {
		pBlk = (request_block_ptr_t) tableGet (pServerData->connectionTable, list[i]);
		if (pBlk != NULL) destroyNamedPipe(pBlk->fullPipeName);
	}		

	free (list);

	return 0;	
}

/**
 *********************************************************
 *
 * @brief
 *    stop the server named pipe. Stop accepting connection requests.
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

int   stopNamedPipeListener( server_block_t* pServerData )
{
	TPTP_LOG_DEBUG_MSG(pServerData, "stopTransportListener (named pipe).") ;

	/* stop the running thread */
	pServerData->threadStatus = IDLE ;

	destroyNamedPipe(pServerData->fullPipeName);
	closeConnections(pServerData);

	return ( 0 ) ;
}

/**
 *********************************************************
 *
 * @brief
 *    terminate the given connection
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

int   terminateNamedPipeConnection(server_block_t* pServerData, tptp_uint32 connectionID)
{
	int     rc = 0 ;

	/* retrieve the corresponding named pipe */
	request_block_ptr_t pBlk = 
		(request_block_ptr_t) tableGet(pServerData->connectionTable, connectionID) ;

	TPTP_LOG_DEBUG_MSG1(pServerData, "terminateConnection (named pipe): connection id(%d).", connectionID) ;

	if (pBlk->connectionType != DATA_CHANNEL)
	{
		/* pass the command to the agent controller to be processed */
		removeConnectionEntry_ptr_t pRemoveFunc = 
			pServerData->agentControllerDataBlk.removeConnectionEntry ;

		pRemoveFunc(pServerData->cmo, pBlk->connectionId) ;
	}
	else
	{
		/* pass the command to the agent controller to be processed */
		removeDataConnectionEntry_ptr_t pDataRemoveFunc = 
			pBlk->pServerData->agentControllerDataBlk.removeDataConnectionEntry ;

		pDataRemoveFunc(pBlk->pServerData->cmo, pBlk->connectionId) ;
	}


	/* stop communicating with the agent */
	/* go close it down */
	cleanPipeUp(&pBlk->agentPipe) ;

	// This call also frees the memory for the pBlk
	tableRemove( pServerData->connectionTable, connectionID );

	return ( rc ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    send a given message to the given destination (identified by connection id)
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
int   sendNamedPipeMessage( server_block_t* pServerData, tptp_uint32 connectionID, tptp_uint32 cmdSize, tptp_string* pCmdBlock)
{
	TPTP_LOG_DEBUG_MSG1(pServerData, "sendMessage (named pipe) to this connection id(%d).", connectionID) ;
	
	return ( sendThisMessage(pServerData, connectionID, cmdSize, pCmdBlock, TRUE ) ) ;
}


/**
 *********************************************************
 *
 * @brief
 *    common interface to set up the data path between source and destination
 *    (named pipe).
 *
 * @return
 *    0 - Success
 *    negative - Not supported
 *    positive - Error number
 *********************************************************/

tptp_int32 setNamedPipeIncomingDataFunc( server_block_t* pServerData, tptp_uint32 connectionID, tptp_uint32 partnerID, tptp_object* partner, sendData_ptr_t newDataFunc )
{
	/* locate the block */
	request_block_ptr_t pBlock = 
		(request_block_ptr_t) tableGet(pServerData->connectionTable, connectionID) ;

	TPTP_LOG_DEBUG_MSG2(pServerData, "setIncomingDataFunc(named pipe) connectionID(%d) partnerID(%d).",
		connectionID, partnerID) ;

	pBlock->connectionPartnerID = partnerID ;
	pBlock->partner = partner;
	pBlock->pSendFunc = newDataFunc ;

	return 0 ;
}


/**
 *********************************************************
 *
 * @brief
 *    send a given message to the given destination (identified by connection id)
 *
 * @return
 *    Negative - Error
 *    0 - Named pipe is closed gracefully
 *    Positive - the number of bytes that have been sent
 *********************************************************/

tptp_int32 sendNamedPipeData(server_block_t* pServerData, tptp_uint32 connectionID, tptp_uint32 dataSize, tptp_string* pDataBlock)
{
	HANDLE  pipeHandle ;

	int bytesSent = 0 ;

	int rc = 0 ;

	/* locate the block */
	request_block_ptr_t pBlock = 
		(request_block_ptr_t) tableGet(pServerData->connectionTable, connectionID) ;

	/* locate the socket to send */
	if (pBlock->consoleInputPipe == 0)
		pipeHandle = pBlock->agentPipe ;
	else
		/* send console data to the process */
		pipeHandle = pBlock->consoleInputPipe ;

	/* go send the message */
	bytesSent = writeToNamedPipe(pipeHandle, pDataBlock, 0, dataSize, &bytesSent);
	if (bytesSent < 0)
	{
		TPTP_LOG_ERROR_MSG(pServerData, "Error: unable to send data to the named pipe.");
	}
	else
	{
		TPTP_LOG_DEBUG_MSG1(pServerData, "Sent out(%d bytes) through named pipe. ", bytesSent) ;
		rc = bytesSent ;
	}
	
	return ( rc ) ;
}



/**
 *********************************************************
 *
 * @brief
 *    process the data received from socket of a specific agent
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
int forwardDataToPartner(request_block_ptr_t pBlk, int dataLen, char *pBuffer)
{
	sendData_ptr_t pSendFunc = pBlk->pSendFunc ;

	TPTP_LOG_DEBUG_MSG2(pBlk->pServerData, "Send data from connectionId(%d) to PartnerID(%d).", 
		pBlk->connectionId, pBlk->connectionPartnerID) ;

	TPTP_LOG_DEBUG_MSG2(pBlk->pServerData, "forwardDataToPartner: Send data length(%d) buffer(%s).", 
		dataLen, pBuffer) ;

	if (pSendFunc != NULL)
		pSendFunc(pBlk->partner, pBlk->connectionPartnerID, dataLen, pBuffer) ;
	else
		TPTP_LOG_ERROR_MSG(pBlk->pServerData, "Data path is NOT established yet.") ;

	return 0 ;
}

/**
 *********************************************************
 *
 * @brief
 *    send data to its "associated" partner with DIME header block
 *
 *********************************************************/

int forwardDataWithDIMEToPartner(request_block_ptr_t pBlk, int dataLen, char *pBuffer, int consoleType)
{
	char *pCopy = NULL ;
	DIME_HEADER_PTR_T dp;	
	int dimeLength = 0;

	#ifdef MVS

		// At this point pBuffer contains NO dime header, and is the exact text is straight from the console (as a char *), so this native2unicode is correct. - jgw

		//Convert Console Channel data
		char * nativeBuffer = 0;
		native2unicode(&nativeBuffer, pBuffer, dataLen);
		pBuffer = nativeBuffer;

		dataLen = strlen(nativeBuffer);
	#endif
	
	dimeLength = MAKE_DIME_CONSOLE(&dp, pBlk->processID4Console, consoleType, dataLen);

	make_dime_header((char *) dp) ;

	/* send the DIME first */
	/* followed by the actual data */
	/* Note: will not work without synchronized socket lock */
	/*
	forwardDataToPartner(pBlk, dimeLength, (char *)dp) ;
	forwardDataToPartner(pBlk, dataLen, pBuffer) ;
	*/
	pCopy = (char *) malloc(dimeLength + dataLen) ;
	memcpy(pCopy, dp, dimeLength) ;

	memcpy(pCopy+dimeLength, pBuffer, dataLen) ;

	#ifdef MVS
		if(nativeBuffer != 0)
			tptp_free(nativeBuffer);
	#endif

	forwardDataToPartner(pBlk, dimeLength+dataLen, pCopy) ;
	tptp_free(dp); dp = NULL;

	return 0 ;
}



