/*******************************************************************************
 * 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:
 *    Vishnu K Naikawadi,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: DataProviderImpl.cpp,v 1.42 2009/08/07 01:35:53 jwest Exp $ 
 *******************************************************************************/ 

#include <stdio.h>
#include <stdlib.h>
#include <iostream>
#ifndef _WIN32
#include <strings.h>
#endif


#include "tptp/agents/DataProviderImpl.h"
#include "tptp/NamedPipeTL.h"
#include "tptp/TPTPSupportUtils.h"
#include "tptp/TPTPOSCalls.h"
#include "tptp/agents/AgentLog.h"

using namespace std;

DataProviderImpl::DataProviderImpl(char* name) : BaseAgentImpl(name)
{
	dataConnectionUUID = NULL;
	serverMemBlockInfo.pRamboBlock = NULL;
	agentMemBlockInfo.pRamboBlock = NULL;
	dataConnMemBlockInfo.pRamboBlock = NULL;
	tptp_initializeLock(&dataConnListLock);
	tptp_list_init( &dataConnectionsList );
}



DataProviderImpl::~DataProviderImpl()
{
	tptp_node_t*  node = NULL;
	tptp_data_conn_t* dataConn = NULL;

	for (node = dataConnectionsList.head; node != 0; node = node->next )
	{
		dataConn = (tptp_data_conn_t*)node->data;

		if (dataConn->_isEstablished) 
		{
			releaseDataConn(dataConn);
		}
	}

	if (dataConnectionUUID) tptp_free(dataConnectionUUID);
	tptp_list_clear( &dataConnectionsList );
	tptp_deleteLock(&dataConnListLock);
}


tptp_data_conn_t* DataProviderImpl::getDataConnection(int clientID)
{
	tptp_node_t*  node = NULL;
	tptp_data_conn_t* dataConn = NULL;
	bool foundConnection = false;

	//Get read lock to the connection list
	tptp_getReadLock(&dataConnListLock);
	for (node = dataConnectionsList.head; node != 0; node = node->next )
	{
		dataConn = (tptp_data_conn_t*)node->data;
		if (dataConn->clientid == clientID)
		{
			foundConnection = true;
			break;
		}
	}
	tptp_releaseReadLock(&dataConnListLock);

	if (!foundConnection)
	{
		dataConn = NULL;
	}

	return dataConn;
}


int DataProviderImpl::sendData(int destinationID, char buffer[], int bufferLength)
{
	//Build the DIME HEADER
	int dimeLength;
	DIME_HEADER_PTR_T dimeHeader;

	dimeHeader = (DIME_HEADER_PTR_T) tptp_malloc(sizeof(DIME_HEADER_T));
	dimeLength = sizeof(DIME_HEADER_T);
	INIT_DIME_HEADER(dimeHeader);
	dimeHeader->data_length = bufferLength;
	MAKE_NBO_DIME_HEADER((char *) dimeHeader);	

	/* send data through the path */
	int rc = this->sendData(destinationID, buffer, bufferLength, dimeHeader, dimeLength);

	tptp_free( dimeHeader );
	
	return rc;

}


int DataProviderImpl::sendData(int destinationID, char buffer[], int bufferLength, DIME_HEADER_PTR_T dimeHeader, int dimeHeaderLength)
{
	tptp_data_conn_t* dataConn = NULL;
	int rc = 0;

	dataConn = getDataConnection(destinationID);

	// The dime header is not a string, we can't just pretend it is!!!!!
//	TPTP_LOG_DEBUG_MSG2(this, "The dime header data - %s length - %d", (char*)dimeHeader, dimeHeaderLength);

	if (dataConn != NULL)
	{
		/* send data through the path */
		rc = ipcMemWriteWithDIME(buffer, bufferLength, (char *) dimeHeader, dimeHeaderLength, &(dataConn->dataConnMemBlockInfo));
	} else {
		/* destinationID not found in the connection list */
		rc = -1;
	}

	return rc;

}


int sharedMemDataPathProcessorFunc(void *pData, int dataLen, void *pArgs)
{
	tptp_data_conn_t* dataConn = (tptp_data_conn_t *) pArgs ;

	DataProviderImpl * pProvider = (DataProviderImpl *) dataConn->pObj ;

//	TPTP_LOG_DEBUG_MSG1(this, "Agent: establishDataPath() receives %d bytes", dataLen) ;

	ipcCloseMem(&dataConn->agentMemBlockInfo);

	if (dataConn->_isEstablished == false)
	{
		// TPTP_LOG_DEBUG_MSG(this, "---> SharedMemDataPathProcessorFunc calls doEstablish()") ;

		pProvider->doEstablish((char *) pData, dataLen, dataConn) ;
		//TODO: Can't say it is truly established at this point since we've not
		//gotten back response from the bind request sent by doEstablish.
		//Can another call to this func happen and we end up in the else clause
		//prior to when the bind call completed?  The dataConnID will be in our table
		//but is it usable?
		dataConn->_isEstablished = true ;
	}
	else
	{
		// TPTP_LOG_DEBUG_MSG(this, "---> SharedMemDataPathProcessorFunc calls receiveData()") ;
		pProvider->processData(dataConn->clientid, (char*)pData, 0, dataLen);
		//pProvider->receiveData(dataConn->clientid, (char*)pData, dataLen);
	}

	return 0 ;
}


int DataProviderImpl::loadMessageHeader(tptp_data_conn_t *ccB, char data[], int offset, int limit) 
{
	/* Load all we can into the header */
	while (offset<limit && ccB->_currentHeaderOffset < (int)sizeof(DIME_HEADER_T))
	{
		ccB->_messageHeader[ccB->_currentHeaderOffset++]=data[offset++];
	}
	if (ccB->_currentHeaderOffset == (int)sizeof(DIME_HEADER_T))
	{
		MAKE_DIME_HEADER(ccB->_messageHeader);
		ccB->_dimeMessageHeader = (DIME_HEADER_PTR_T)ccB->_messageHeader;
	}	

	return offset;
}

int DataProviderImpl::loadMessageHeaderDetails(tptp_data_conn_t *ccB, char data[], int offset, int limit) 
{
	/* Load all we can into the header */
	while (offset<limit && ccB->_currentHeaderOffset < (int)(sizeof(DIME_HEADER_T)+ccB->_dimeMessageHeader->id_length+ccB->_dimeMessageHeader->options_length+ccB->_dimeMessageHeader->type_length))
	{
		ccB->_messageHeader[ccB->_currentHeaderOffset++]=data[offset++];
	}
	if (ccB->_currentHeaderOffset == (int)(sizeof(DIME_HEADER_T)+ccB->_dimeMessageHeader->id_length+ccB->_dimeMessageHeader->options_length+ccB->_dimeMessageHeader->type_length))
	{
		ccB->_dimeMessageHeader = (DIME_HEADER_PTR_T)ccB->_messageHeader;
	}	

	return offset;
}

long DataProviderImpl::getMessageLength(tptp_data_conn_t *ccB) 
{	
	TPTP_LOG_DEBUG_MSG1(this, "The data length is - %d", ccB->_dimeMessageHeader->data_length);
	return ccB->_dimeMessageHeader->data_length;
}

int DataProviderImpl::processData(int clientid, char data[], int offset, int limit)
{
	long messageLength=0;
	int current;

	tptp_data_conn_t* ccB = getDataConnection(clientid);

	if (!ccB)
	{
		// Data connection not found for client id
		TPTP_LOG_ERROR_MSG1(this, "DataProviderImpl: processData, failed to find data connection for client ID %d", clientid);
		return -1;
	}

	current=offset;

	if ( offset>=limit ) {
	   return limit;
	}

	while(current<limit) 
	{
		/* Is this a new message? */
		if(ccB->_currentHeaderOffset<(int)sizeof(DIME_HEADER_T)) {
			/* Load the message header */
			current=this->loadMessageHeader(ccB, data, current, limit);

			/* Did we get the entire header, if not return */
			if(current==limit) {
				return current;
			}		
		}
	
		if(ccB->_currentHeaderOffset < (int)(sizeof(DIME_HEADER_T)+ccB->_dimeMessageHeader->id_length+ccB->_dimeMessageHeader->options_length+ccB->_dimeMessageHeader->type_length)) 
		{
			/* Load the message header */
			current=this->loadMessageHeaderDetails(ccB, data, current, limit);

			/* Did we get the entire header, if not return */
			if (current==limit && ccB->_currentHeaderOffset < (int)(sizeof(DIME_HEADER_T)+ccB->_dimeMessageHeader->id_length+ccB->_dimeMessageHeader->options_length+ccB->_dimeMessageHeader->type_length))
			{
				return current;
			}
			if(ccB->_dimeMessageHeader->data_length > 0 && current==limit) {
				return current;
			}		
		}	 

		//TODO - Check if the message is valid	

		/* How long is the current message */
		messageLength=getMessageLength(ccB);

		/* Copy as many bytes as possible into the forwarding buffer */
		if (ccB->_dimeMessageHeader->data_length > 0)
		{
			while(current<limit && ccB->_bytesWritten<(int)(ccB->_dimeMessageHeader->data_length)) {
				ccB->_binaryForwardBuffer[ccB->_bytesWritten++]=data[current++];
			}
		}
		/* Are we at the end of the message? If so forward to the handler */
		if(ccB->_bytesWritten==(int)(ccB->_dimeMessageHeader->data_length)) 
		{
			TPTP_LOG_DEBUG_MSG1(this, "Calling receive data - %d", ccB->_bytesWritten);
			this->receiveData(clientid, ccB->_binaryForwardBuffer, ccB->_bytesWritten, ccB->_dimeMessageHeader);
			ccB->_bytesWritten=0;
			ccB->_currentHeaderOffset=0;
		}	
	}

	return current;
}

int DataProviderImpl::doEstablish(char *buffer, int dataLen, tptp_data_conn_t *dataConn)
{
	char   bindConnections[] = "<Cmd src=\"%d\" dest=\"%d\" ctxt=\"%d\"> <bindDataConnections iid=\"org.eclipse.tptp.agentManager\"><dataConnection1>%ld</dataConnection1><dataConnection2>%ld</dataConnection2></bindDataConnections></Cmd>";

	char command[COMMAND_BUFFER_LENGTH];

	int dataConnectionId2 = -1;

	char *pCurr = buffer ;
	//tptp_basic_msg_header_ptr_t pMsg = (tptp_basic_msg_header_ptr_t) buffer ;

	// read in the connection id //
	pCurr = pCurr + sizeof(tptp_basic_msg_header_t) ;
	pCurr = (char*)readUINTFromBuffer((unsigned char*)pCurr, (unsigned int*)&dataConnectionId2);
	dataConn->agentDataConnID = dataConnectionId2;
 			
	unsigned int dataConnUUIDLength = 0;
	pCurr = (char*)readUINTFromBuffer((unsigned char*)pCurr, (unsigned int*)&dataConnUUIDLength);

	// TODO: Make dataConnUUID in tptp_data_conn_t dynamicly allocated too, so that the following error check can be deleted.
	if (sizeof(dataConn->dataConnUUID) <= dataConnUUIDLength)
	{
		TPTP_LOG_ERROR_MSG(this, "DataProviderImpl: Internal mem err, failed to establish data connection");
		return -1;
	}
 			
	// save away the shared memory name 
	memcpy(dataConn->dataConnUUID, pCurr, dataConnUUIDLength);
	dataConn->dataConnUUID[dataConnUUIDLength] = '\0';
 			
	dataConn->_currentHeaderOffset = 0;
	dataConn->_currentBufferSize = 0;
	dataConn->_bytesWritten = 0;

	//Getting a write lock to avoid the multiple threads accessing the list at the same time
	tptp_getWriteLock(&dataConnListLock);
	tptp_list_add(&dataConnectionsList, (void*)dataConn); //TODO: Check for error return
	tptp_releaseWriteLock(&dataConnListLock);

	//send bindDataConnections
	sprintf( command, bindConnections, getAgentID(), getAgentControllerID(),
		dataConn->bindContextID, dataConn->clientDataConnID, dataConnectionId2);

	TPTP_LOG_DEBUG_MSG(this, "DataProviderImpl: Sending bindConnections Command...");
	
	sendCommand(command);	

	return 0 ;
}

int DataProviderImpl::purgeConnectionList()
{

	tptp_node_t*  node;
	tptp_data_conn_t*      dataConn;

	tptp_getWriteLock(&dataConnListLock);
	node = dataConnectionsList.head;
	while ( node != 0 )
	{
		dataConn = (tptp_data_conn_t*)node->data;

		/* Increment this here, because we might be about to delete the current node */
		node = node->next;

		/* If the agent process is no longer active, get rid of this entry */
		if ( dataConn->_flushingFinished)
		{
			tptp_list_remove( &dataConnectionsList, dataConn );
		}
	}
	tptp_releaseWriteLock(&dataConnListLock);
	return 0;
}

THREAD_USER_FUNC_RET_TYPE DataProviderImpl::handleSharedMemMsg(void* pRequestBlock)
{
	tptp_data_conn_t* dataConn = (tptp_data_conn_t *) pRequestBlock ;
	mem_map_block_t *mptr = NULL;

	if (dataConn->agentConnUUID != NULL) { 
		/* attach to this block so that it will not be deleted */
		mptr = (mem_map_block_t *) malloc(sizeof(mem_map_block_t));
		ipcMemOpen (dataConn->agentConnUUID, mptr);
	}

	// Analysis of lowest level that makes the actual call to the user-provided data processing function
	// (sharedMemDataPathProcessorFunc in this case) shows that specific error codes returned by that
	// function are replaced by a simple -1.
	int rc = ipcFlushToFunc(&(dataConn->agentMemBlockInfo), sharedMemDataPathProcessorFunc, pRequestBlock) ;

	dataConn->_isEstablished = false;
	dataConn->_flushingFinished = true;

	if (mptr != NULL) {
		ipcMemDetach (mptr);
		free (mptr);
	}

	ipcCloseMem(&dataConn->agentMemBlockInfo);
	ipcMemDestroy(&dataConn->agentMemBlockInfo);

	if (rc < 0)
	{
		//TODO: This thread needs to report failure to process data on the shared memory connection.
	}
	
	((DataProviderImpl *) dataConn->pObj)->purgeConnectionList();

	return 0 ;
}

int DataProviderImpl::establishDataPath(int dataConnectionID, int dataChannelType, int destinationID, int contextID)
{
	int rc = -1;
	int dataConnectionId1 = dataConnectionID;
	int flags = dataChannelType;

	long dataBindCmdContextId = this->getNextContext();

	tptp_data_conn_t* dataConn;	

	purgeConnectionList();

	TPTP_LOG_DEBUG_MSG1(this, "DataProviderImpl: establishDataPath() called, connID %d", dataConnectionID);

	dataConn = (tptp_data_conn_t*)tptp_malloc( sizeof(tptp_data_conn_t) );
	if ( dataConn == 0 )
	{
		/* TODO: Report an error */
		return -1;
	}

	dataConn->clientid = destinationID;
	dataConn->contextID = contextID;
	dataConn->bindContextID = dataBindCmdContextId;
	dataConn->clientDataConnID = dataConnectionId1;
	dataConn->type = dataChannelType;
	dataConn->pObj = (void *) this ;
	dataConn->_isEstablished = false ;
	dataConn->_isReleased = false ;
	dataConn->_flushingFinished = false ;

	if (!isMemOpen(&serverMemBlockInfo)) {
		rc = ipcMemOpen(_raShmBufNameRoot, &serverMemBlockInfo) ;
		if (rc != 0)
			return -2 ;
	}

	// generate the unique id string 
	// This unique id is based on the shm config name combined with the
	// client id, agent id and local number of data path.
	// This id must be less than 24 to be compatible with the RAC.

	static int shmBufNum = 0;
	char *pUniqueId = dataConn->agentConnUUID;
	sprintf(pUniqueId,"%s_%d_%d_%d", _raShmBufNameRoot, destinationID, getAgentID(), ((shmBufNum++)%100));

	// create the agent's shared memory 
	rc = ipcMemCreate(pUniqueId, &(dataConn->agentMemBlockInfo), 1); // This gets the minimum size.  We don't actually use it.
	if (rc != 0)
		return -2 ;

	// go send the CONNECT command 
	rc = sendCONNECT_DATACommand(pUniqueId, flags) ;
	if (rc == 0)
	{
		TID tid;
		HANDLE    th ;

		tptpStartThread(handleSharedMemMsg, (void *) dataConn, &tid, &th);

		CLOSE_THREAD_HANDLE(th);
	}

	return rc;
}

int DataProviderImpl::releaseDataPath(int dataConnectionID)
{
	tptp_node_t*  node = NULL;
	tptp_data_conn_t* dataConn = NULL;
	bool foundConnection = false;

	purgeConnectionList();

	tptp_getReadLock(&dataConnListLock);
	for (node = dataConnectionsList.head; node != 0; node = node->next )
	{
		dataConn = (tptp_data_conn_t*)node->data;
		if (dataConn->clientDataConnID == dataConnectionID)
		{
			foundConnection = true;
			break;
		}
	}
	//Should we also be removing the connection from the list.
	//We should investiagte this.
	tptp_releaseReadLock(&dataConnListLock);

	if (foundConnection)
	{
		releaseDataConn (dataConn);
	} else {
		TPTP_LOG_ERROR_MSG1(this, "Error: did not find connection %d to release ", dataConnectionID);
	}

	return 0;
}

int DataProviderImpl::releaseDataConn(tptp_data_conn_t* dataConn)
{
	if (dataConn != NULL) {
		//We can only released once.
		//Our code handles release by explicit call, destructor and notify of dereference.
		if (!dataConn->_isReleased) 
		{
		//Call stop flushing on the AC shared memory block.
			dataConn->_isReleased = true;
			
			ipcCloseMem(&dataConn->dataConnMemBlockInfo);
			ipcMemDetach(&dataConn->dataConnMemBlockInfo);

			ipcCloseMem(&dataConn->agentMemBlockInfo);
			ipcStopFlusher(&dataConn->agentMemBlockInfo);
			
			sendDisconnectCommand (dataConn);
		}
	}

	return 0;
}

int DataProviderImpl::connectionsBound(int contextID)
{

	tptp_node_t*  node = NULL;
	tptp_data_conn_t* dataConn = NULL;

	//Get read lock to the connection list
	tptp_getReadLock(&dataConnListLock);
	for (node = dataConnectionsList.head; node != 0; node = node->next )
	{
		dataConn = (tptp_data_conn_t*)node->data;
		if (dataConn->bindContextID == contextID)
		{
			break;
		}
	}
	tptp_releaseReadLock(&dataConnListLock);

	//Open the connection
	ipcMemOpen(dataConn->dataConnUUID, &(dataConn->dataConnMemBlockInfo));


	//Send the Response to the Client
	char   commandFormat[] = "<Cmd src=\"%ld\" dest=\"%ld\" ctxt=\"%ld\"> <dataPathEstablished iid=\"org.eclipse.tptp.Client\"></dataPathEstablished></Cmd>";
	char   command[COMMAND_BUFFER_LENGTH];
	sprintf( command, commandFormat, getAgentID(), dataConn->clientid, dataConn->contextID);

	
	sendCommand(command);

	return 0;
}



int DataProviderImpl::connectionsBindError(int contextID, int error)
{


	tptp_node_t*  node = NULL;
	tptp_data_conn_t* dataConn = NULL;

	//Get read lock to the connection list
	tptp_getReadLock(&dataConnListLock);
	for (node = dataConnectionsList.head; node != 0; node = node->next )
	{
		dataConn = (tptp_data_conn_t*)node->data;
		if (dataConn->contextID == contextID)
		{
			break;
		}
	}
	tptp_releaseReadLock(&dataConnListLock);

	//Send the Response to the Client
	char   commandFormat[] = "<Cmd src=\"%ld\" dest=\"%ld\" ctxt=\"%ld\"> <dataPathNotEstablished iid=\"org.eclipse.tptp.Client\"><reason>%d</reason></dataPathNotEstablished></Cmd>";
	char   command[COMMAND_BUFFER_LENGTH];
	sprintf( command, commandFormat, getAgentID(), dataConn->clientid, dataConn->contextID, error);

	
	sendCommand(command);

	return 0;
}



int DataProviderImpl::processDataProviderCommands(CmdBlock* cmd)
{
	int ret = -1;

	char* cmdName = cmd->getCommandName();

	TPTP_LOG_DEBUG_MSG3(this, "DataProviderImpl: processDataProviderCommands() called. Command name - %s, source ID - %d, context ID - %d", cmdName, cmd->getSourceID(), cmd->getContextID());

	if (isEqualString(cmdName, "establishDataPath"))
	{
		int dataConnID;
		if (0 != getIntegerParam("dataConnectionID", cmd->getParamList(), &dataConnID))
		{
			TPTP_LOG_DEBUG_MSG1(this,"processDataProviderCommands: cmd %s missing dataConnectionID param", cmdName);
			return -1;
		}

		int dataChannelType;
		if (0 != getIntegerParam("flags", cmd->getParamList(), &dataChannelType))
		{
			TPTP_LOG_DEBUG_MSG1(this,"processDataProviderCommands: cmd %s missing flags param", cmdName);
			return -1;
		}

		ret = establishDataPath(dataConnID, dataChannelType, cmd->getSourceID(), cmd->getContextID());				
	    TPTP_LOG_DEBUG_MSG1(this, "DataProviderImpl: Establish DataPath returned result %d", ret);
	}
	else if (isEqualString(cmdName, "dataConnectionsBound"))
	{
		// connections bound
		ret = connectionsBound(cmd->getContextID());		
	    TPTP_LOG_DEBUG_MSG1(this, "DataProviderImpl: dataConnections bound returned result %d", ret);
	}
	else if (isEqualString(cmdName, "dataConnectionBindingFailed"))
	{
		// TODO: Can't find code that generates this failure, so cannot be
		// sure what the parameter name is.  Assuming it is "error" for now.

		// connections not bound
		int errorVal;
		if (0 != getIntegerParam("error", cmd->getParamList(), &errorVal))
		{
			TPTP_LOG_DEBUG_MSG1(this,"processDataProviderCommands: cmd %s missing error param", cmdName);
			return -1;
		}
		ret = connectionsBindError(cmd->getContextID(), errorVal);		
	    TPTP_LOG_DEBUG_MSG1(this, "DataProviderImpl: dataConnectionBindingFailed, returned result %d", ret);
	}
	else if (isEqualString(cmdName, "releaseDataPath"))
	{
		int dataConnID;

		if (0 != getIntegerParam("dataConnectionID", cmd->getParamList(), &dataConnID))
		{
			TPTP_LOG_DEBUG_MSG1(this,"processDataProviderCommands: cmd %s missing dataConnectionID param", cmdName);
			return -1;
		}
		ret = this->releaseDataPath(dataConnID);

	        TPTP_LOG_DEBUG_MSG(this, "DataProviderImpl: releaseDataPath ");
	}

	return ret;

}


/**
 *********************************************************
 *                                                       
 * @brief                                                
 *    send in the CONNECT command
 *
 * @return
 *    0 if success
 *    nonzero if error
 *
 *********************************************************/
int DataProviderImpl::sendCONNECT_DATACommand(char *pUniqueId, int flags)
{
	int ret = 0 ;

	char *buffer = NULL;
	int  bufferLength = 0;
	//int  bytesSent = 0 ;			
	int  dataPathType;
	int  uuidLength;
	char *cmd = NULL;
	int  cmdLength = 0;
	char *pCurr = NULL ;

	/* Connect Data cmd contains: dataPathType + uuidLength + uuid string */
	cmdLength = sizeof(int) + sizeof(int) + strlen(pUniqueId) + 1;
	cmd = (char*)tptp_malloc(cmdLength);
	if (!cmd) return -1;

	// prepare the data 
	dataPathType = flags;

	pCurr = cmd ;
	pCurr = (char*)writeUINTToBuffer((unsigned char*)pCurr, dataPathType);

	uuidLength = strlen(pUniqueId) ;
	pCurr = (char*)writeUINTToBuffer((unsigned char*)pCurr, uuidLength);

	strcpy(pCurr, pUniqueId) ;

	// build the CONNECT command string 
	flags = flags | CONNECT_DATA ;
	addBasicMsgHeader(cmd, cmdLength, &buffer, &bufferLength, flags) ;

	//TPTP_LOG_DEBUG_MSG2(this, "DataProviderImpl:uuidLength(%d) bufferLength(%d)", strlen(pCurr), bufferLength);

	// send a message to the agent controller (server) 
	ret = ipcMemWrite(buffer, bufferLength, &serverMemBlockInfo);
	if (ret != 0)
	{
		TPTP_LOG_ERROR_MSG(this, "DataProviderImpl: Error - unable to send data to the shared memory block.");
	}
	else
	{
		printThisEnvelope((tptp_basic_msg_header_ptr_t) buffer) ;
	}

	if (cmd) tptp_free(cmd);
	if (buffer) tptp_free(buffer);
	return ( ret ) ;

}

int DataProviderImpl::sendDisconnectCommand(tptp_data_conn_t* dataConn) {
	char *buffer = NULL;
	int  bufferLength = 0;
    char cmd[4];
	int ret = 0 ;
	
	if (dataConn == NULL) return -1;

	writeUINTToBuffer((unsigned char*)cmd, dataConn->agentDataConnID);
	addBasicMsgHeader(cmd, sizeof(int), &buffer, &bufferLength, DISCONNECT) ;

	ret = ipcMemWrite(buffer, bufferLength, &serverMemBlockInfo);
	if (ret != 0) {
		TPTP_LOG_ERROR_MSG(this, "DataProviderImpl: Error - unable to send DISCONNECT request to the shared memory block.");
	}

	if (buffer) tptp_free(buffer);
	
	return ret;
}

//This method is overridding the base method in BaseAgentImpl
//The intent is to call the Base removeClient and then call the releaseDataPath
//if the data connection has been established.
void DataProviderImpl::removeClient(tptp_int32 clientID)
{
	BaseAgentImpl::removeClient(clientID);

	tptp_data_conn_t* dataConn = getDataConnection(clientID);

	if (dataConn == NULL) 
	{
		TPTP_LOG_ERROR_MSG(this, "DataProviderImpl: removeClient - no data connection for client.");
		return;
	}

	if (dataConn->_isEstablished) 
	{
		releaseDataPath(dataConn->clientDataConnID);
	}
}
