/*******************************************************************************
 * Copyright (c) 2005, 2009 Intel Corporation and others.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v1.0
 * which accompanies this distribution, and is available at
 * http://www.eclipse.org/legal/epl-v10.html
 *
 * Contributors:
 *    Andy Kaylor, Intel - Initial API and Implementation
 *    Spundun Bhatt (spundun@gmail.com), with the support and encouragement of the University of Southern California Information Sciences Institute Distributed Scalable Systems Division.
 *
 * $Id: AgentCTL.c,v 1.28 2009/08/07 01:35:53 jwest Exp $
 *
 *******************************************************************************/ 

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

#include <stdio.h>
#ifdef _WIN32
	#include <io.h>
#else
	#include "tptp/TPTPCommon.h"
#endif

/**
 *********************************************************
 *
 * @brief
 *    common interface to instantiate a transport layer object
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

tptp_int32 createTransportListener( tptp_object* cmo, transport_layer_data_t * pTransportData, tptp_object* tlo )
{
	tptp_int32 rc;
	
	rc = baseTL_createTransportListener( cmo, pTransportData, tlo, "Agent Compatibility TL" );
	if ( ( rc == 0 ) && (tlo != NULL) && (tlo->data != NULL) )
	{
		actl_state_data_t*  actlData;
		tl_state_data_t*    stateData = (tl_state_data_t*)tlo->data;

		/* Allocate implementation specific state data */
		stateData->implData = tptp_malloc( sizeof(actl_state_data_t) );
		if ( stateData->implData == NULL )
		{
			baseTL_destroyTransportListener( tlo );
			return TPTP_ERROR_OUT_OF_MEMORY;
		}
		actlData = (actl_state_data_t*)stateData->implData;

		/* Save the master pipe name */
		strcpy( actlData->fullPipeName, RAC_PIPE_NAMESPACE );
		strcat( actlData->fullPipeName, RAC_MASTER_ADDRESS );

		/* Set up the process list */
		ra_initializeDataStructures();
		actlData->processList = ra_getActiveProcessList();

		/* Start a thread to clear our process list as needed */
		startProcessPurgeThread(stateData);
	}

	return 0;
}


/**
 *********************************************************
 *
 * @brief
 *    common interface to send to the given destination the given message
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

tptp_int32 sendMessage( tptp_object* tlo, tptp_uint32 connectionID, tptp_uint32 size, tptp_string* pCmdBlock)
{
	tptp_int32                ret;
	tl_state_data_t*          stateData;
	tl_control_connection_t*  connectionEntry;
	agent_t*                  agent;
	actl_state_data_t*        actlData;

	if ( !isValidTPTPBlock(tlo, TPTP_BASETL_MAKE_OBJECT_ID(tlo)) )
		return TPTP_ERROR_INVALID_PARAMETER;

	stateData = tlo->data; /* This was verified not to be NULL by isValidTPTPBlock */

	actlData  = (actl_state_data_t*)stateData->implData;
	if ( actlData == NULL )
	{
		/* This would be very unexpected */
		TPTP_LOG_ERROR_MSG( stateData, "ACTL cannot find TL-specific state data in sendMessage" );
		return -1;
	}

	/* retrieve the corresponding agent information */
	connectionEntry = (tl_control_connection_t*) tableGet(stateData->controlConnectionTable, connectionID) ;
	if ( connectionEntry == NULL )
	{
		/* Unrecognized connection */
		TPTP_LOG_ERROR_MSG1( stateData, "ACTL cannot find connection entry for ID %d", connectionID );
		return -1;
	}

	ra_mutexEnter( &actlData->processList->lock );

	agent = (agent_t*)connectionEntry->implData;
	if ( agent == NULL )
	{
		/* Missing agent data */
		ra_mutexExit( &actlData->processList->lock );
		TPTP_LOG_ERROR_MSG1( stateData, "ACTL cannot find agent data for ID %d", connectionID );
		return -1;
	}

	/* We can't hold the lock while we're processing this, but we don't want
	   the agent to be purged.  This flag will tell the purge thread to wait */
	actlData->agentInUse++;
	ra_mutexExit( &actlData->processList->lock );

	/* Call our handler to parse, translate and process this message */
	ret = handleOutgoingMessage( stateData, agent, pCmdBlock );

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

	return ret;
}


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

tptp_int32 terminateConnection(tptp_object* tlo, tptp_uint32  connectionID)
{
	tl_state_data_t*          stateData;
	tl_control_connection_t*  connectionEntry;
	agent_t*                  agent;

	if ( !isValidTPTPBlock(tlo, TPTP_BASETL_MAKE_OBJECT_ID(tlo)) )
		return TPTP_ERROR_INVALID_PARAMETER;

	stateData = tlo->data; /* This was verified not to be NULL by isValidTPTPBlock */

	/* retrieve the corresponding agent information */
	connectionEntry = (tl_control_connection_t*) tableGet(stateData->controlConnectionTable, connectionID) ;
	if ( connectionEntry == NULL )
	{
		/* Unrecognized connection */
		/* TODO: Log an error */
		return -1;
	}
	agent = (agent_t*)connectionEntry->implData;
	if ( agent == NULL )
	{
		/* Missing agent data */
		/* TODO: Log an error */
		return -1;
	}

	return 0;
}


/**
 *********************************************************
 *
 * @brief
 *    common interface to send data to the given destination 
 *
 * @return
 *    0 - Success
 *    negative - Not supported
 *********************************************************/

tptp_int32 sendData( tptp_object* tlo, tptp_uint32 connectionID, tptp_uint32 size, void * pDataBlock)
{
	tl_state_data_t* stateData;

	if ( !isValidTPTPBlock(tlo, TPTP_BASETL_MAKE_OBJECT_ID(tlo)) )
		return TPTP_ERROR_INVALID_PARAMETER;

	stateData = tlo->data; /* This was verified not to be NULL by isValidTPTPBlock */

	// TODO: NOT_SUPPORTED?

	return 0;
}


/**
 *********************************************************
 *
 * @brief
 *    common interface to terminate a given data connection
 *
 * @return
 *    0 - Success
 *    negative - Not supported
 *********************************************************/

tptp_int32 terminateDataConnection(tptp_object* tlo, tptp_uint32  connectionID)
{
	tl_state_data_t* stateData;
	tl_data_connection_t*  connectionEntry;

	if ( !isValidTPTPBlock(tlo, TPTP_BASETL_MAKE_OBJECT_ID(tlo)) )
		return TPTP_ERROR_INVALID_PARAMETER;

	stateData = tlo->data; 

	connectionEntry = (tl_data_connection_t*) tableGet(stateData->dataConnectionTable, connectionID) ;
	if ( connectionEntry == NULL )
	{
		return -1;
	}

	/*TODO */

	return 0;
}




/**
 *********************************************************
 *
 * @brief
 *    common interface to destroy a transport layer object
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
tptp_int32 destroyTransportListener( tptp_object* tlo )
{
	return baseTL_destroyTransportListener( tlo );
}


/**
 *********************************************************
 *
 * @brief
 *    common interface to start an underlying transport mechanism
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

tptp_int32 startTransportListener(tptp_object* tlo)
{
	HANDLE              handle;
	actl_state_data_t*  actlData;
	tl_state_data_t*    stateData;
	
	if ( !isValidTPTPBlock(tlo, TPTP_BASETL_MAKE_OBJECT_ID(tlo)) )
		return TPTP_ERROR_INVALID_PARAMETER;

	stateData = (tl_state_data_t*)tlo->data;

	actlData = (actl_state_data_t*)stateData->implData;
	if ( actlData == NULL )
		return -1;

	/* Create the pipe on which we will listen for connections */
	handle = createReadOnlyNamedPipe(RAC_PIPE_NAMESPACE, RAC_MASTER_ADDRESS, FALSE);
	if (handle == INVALID_HANDLE_VALUE)
	{
		TPTP_LOG_SEVERE_MSG(stateData, "ERROR: Cannot start Agent Controller. ");
		TPTP_LOG_SEVERE_MSG1(stateData,"REASON: Cannot open named pipe %s. Make sure the named pipe is free and no other Agent Controller is running or any LOCAL direct connections exist.", RAC_MASTER_ADDRESS );
		baseTL_destroyTransportListener( tlo );
		return -1;
	}
	setHandleInherited(handle);

	actlData->serverPipe = handle;
	
	return baseTL_startTransportListener( tlo, serveRequests );
}

int clearProcesses() {
	process_list_t *plist;
	process_list_node_t *pnode;
	process_list_node_t *nexttemp;

	plist = ra_getActiveProcessList();
	if (plist == NULL) return 0;

	pnode = plist->head;

	while(pnode != NULL) {

		nexttemp = pnode->next;

		ra_processDestroy(pnode->process);

		pnode = nexttemp;
	}

	return 0;
}

/**
 *********************************************************
 *
 * @brief
 *    common interface to stop the transport mechanism
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/

tptp_int32 stopTransportListener(tptp_object* tlo) {
	tptp_int32 rc;
	char pipeName[_MAX_PATH];

	rc = baseTL_stopTransportListener(tlo);

	strcpy (pipeName, RAC_PIPE_NAMESPACE);
	strcat (pipeName, RAC_MASTER_ADDRESS);

	destroyNamedPipe(pipeName);	
	clearProcesses();

	return rc;
}


/**
 *********************************************************
 *
 * @brief
 *    common interface to set the function and data block 
 * for forwarding messages
 *
 * @return
 *    0 - Success
 *    nonzero - Error.
 *********************************************************/
tptp_int32 setProcessMessageFunc(tptp_object* tlo, tptp_object* nexto, processMessage_ptr_t func )
{
	return baseTL_setProcessMessageFunc( tlo, nexto, func );
}

/**
 *********************************************************
 *
 * @brief
 *    common interface to set up the data path between source and destination
 *
 * @return
 *    0 - Success
 *    negative - Not supported
 *********************************************************/

tptp_int32 setIncomingDataFunc( tptp_object* tlo, tptp_uint32 connectionID, tptp_uint32 partnerID, tptp_object* partner, sendData_ptr_t newDataFunc )
{
	return baseTL_setIncomingDataFunc( tlo, connectionID, partnerID, partner, newDataFunc );
}

int processFileData(void *pTraceRecord, int recordLength, void *pArgs) {
	int bytesWritten = 0;
	char cr = '\n';

	if(pArgs  && recordLength>0 && pTraceRecord) {
		int fd=(int)pArgs;
		bytesWritten = write(fd, (char*)pTraceRecord+10, recordLength-10);
		write(fd, &cr, 1);

		if ( (int)-1 == bytesWritten ) {
			//TPTP_LOG_ERROR_MSG2( clientInfo->stateData, "processFileData:  failed writing part of agent %s's shared memory buffer to file. errno=%d", pArgs, errno );
			return -1;
		}
	}
	return 0;
}

THREAD_USER_FUNC_RET_TYPE doFlushToFD(LPVOID args) 
{
	tl_data_connection_t*         connectionEntry = (tl_data_connection_t*)args;
	actl_data_connection_block_t* dataInfo;
	mem_map_block_ptr_t           pMemBlockInfo;
	Lock_t*                       pLock;
	int 	 					  nconnect = CONNECT_TIMEOUT / FLUSHER_TIMEOUT;

	dataInfo = (actl_data_connection_block_t*) connectionEntry->implData;

	pMemBlockInfo = dataInfo->memBlock;

	if ( (dataInfo->flags & TPTP_DATA_PATH_DIRECT_SOCKET) )
	{
		while (nconnect-- > 0) {
			/* This call actually makes all subsequent data transfer happen */
			/* Note the 2 established this as a socket transfer             */
			int result = 0;
			result = ipcFlushToFD(pMemBlockInfo, connectionEntry->dataPartner.descriptor, 2);
			if (result != OSS_ERR_TIMEOUT) break;
		}
		 
		closeThisSocket(connectionEntry->dataPartner.descriptor);
	} else {
		ipcFlushToFunc(pMemBlockInfo, processFileData, (void *)  connectionEntry->dataPartner.descriptor);
		close(connectionEntry->dataPartner.descriptor);
	}

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

	/* Destroy the shared memory resource and free our connection data */
	ipcMemDestroy(dataInfo->memBlock);
	dataInfo->stateData->ac.removeDataConnection( dataInfo->stateData->ac.cmo, connectionEntry->connectionID );
	baseTL_removeDataConnectionEntry( dataInfo->stateData, connectionEntry->connectionID );

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

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

/**
 *********************************************************
 *
 * @brief
 *    common interface to set up the data path between source and destination
 *
 * @return
 *    0 - Success
 *    negative - Not supported
 *********************************************************/

tptp_int32 setDataDescriptor( tptp_object* tlo, tptp_uint32 connectionID, tptp_uint32 partnerID, tptp_object* partner, tptp_uint32 descriptor )
{
 	TID threadId;
	HANDLE threadHandle;
	tl_state_data_t*              stateData;
	tl_data_connection_t*         connectionEntry;
	actl_data_connection_block_t* dataInfo;
	mem_map_block_ptr_t           pMemBlockInfo;

	if ( !isValidTPTPBlock(tlo, TPTP_BASETL_MAKE_OBJECT_ID(tlo)) )
		return TPTP_ERROR_INVALID_PARAMETER;

	stateData = tlo->data; /* This was verified not to be NULL by isValidTPTPBlock */

	/* retrieve the corresponding client information */
	connectionEntry = (tl_data_connection_t*) tableGet(stateData->dataConnectionTable, connectionID) ;
	if ( connectionEntry == NULL )
	{
		/* Unrecognized connection */
		/* TODO: Log an error */
		return -1;
	}

	dataInfo = (actl_data_connection_block_t*) connectionEntry->implData;

	pMemBlockInfo = dataInfo->memBlock;

	/* I don't think we need this information, but now is the last time we'll get it
	   Therefore, I'm saving it just in case we need it in the future */
	connectionEntry->dataPartner.connectionID = partnerID;
	connectionEntry->dataPartner.nexto = partner;
	connectionEntry->dataPartner.sendData = NULL;
	connectionEntry->dataPartner.descriptor = descriptor;


	tptpStartThread(doFlushToFD, (LPVOID) connectionEntry, &threadId, &threadHandle);
	CLOSE_THREAD_HANDLE(threadHandle);

	return 0;
}

