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


#include "tptp/TPTPUtils.h"
#include "tptp/ACFlags.h"
#include "ConnectionManagerLog.h"
#include "AgentManager.h"
#include "ConnectionManager.h"
#include "ConnectionManagerPrivate.h"
#include "tptp/TPTPCommon.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#ifdef _AIX
	#include <errno.h>
	#include <dlfcn.h>
#endif

tptp_int32 connectionManager_init( ConnectionManager_t* cm )
{
	tptp_list_init( &(cm->messagePipelineList) );
	tptp_list_init( &(cm->connectionList) );
	tptp_list_init( &(cm->dataConnectionList) );

	cm->nextTransportID = 1000;
	cm->nextConnectionID = 100;
	cm->nextDataConnectionID = 5000;

	cm->cmo.objectID = CONNECTION_MANAGER;
	cm->cmo.data = (void*)cm;

	return 0;
}

tptp_int32 connectionManager_cleanup( ConnectionManager_t* cm )
{
	connectionManager_stopServers( cm  );

	tptp_list_clear( &(cm->messagePipelineList) );
	tptp_list_clear( &(cm->connectionList) );
	tptp_list_clear( &(cm->dataConnectionList) );

	return 0;
}

tptp_int32 connectionManager_createMessagePipeline( ConnectionManager_t* cm,
                                                    const char*          transportLayer,
                                                    const char*          transportType,
                                                    const char*          tlConfigBlock,
                                                    const char*          commandExtractor )
{
	message_pipeline_t*  mp;
	tptp_int32           ret = 0;

	/* Allocate a block to hold our results */
	mp = (message_pipeline_t*)tptp_malloc( sizeof(message_pipeline_t) );
	if ( 0 == mp )
	{
		/* Log an error */
		TPTP_LOG_ERROR_MSG( cm, "Out of memory when while creating message pipeline." );
		return -1;
	}

	/* Load the transport layer */
	ret = connectionManager_loadTransportLayer( cm, transportLayer, tlConfigBlock, &mp->tl );
	if ( ret != 0 )
	{
		/* Errors should have been logged in the function call */
		return ret;
	}
	else
	{
		if ( transportType != NULL )
		{
			mp->tl.type = (tptp_string*)tptp_malloc(strlen(transportType)+1);
			strcpy( mp->tl.type, transportType );
		}
		else
		{
			mp->tl.type = NULL;
		}
	}

	/* Load the command extractor */
	ret = connectionManager_loadCommandExtractor( cm, commandExtractor, &mp->cx );
	if ( ret != 0 )
	{
		/* Errors should have been logged in the function call */
		return ret;
	}

	/* Hook these two together */
	ret = mp->tl.setProcessMessageFunc( &(mp->tl.tlo), &(mp->cx.cxo), mp->cx.processMessage );
	if ( ret != 0 )
	{
		/* Log an error */
		TPTP_LOG_WARNING_MSG2( cm, "Error received from %s::setProcessMessageFunc, %d", transportLayer, ret );
		return ret;
	}

	/* add this to our list */
	tptp_list_add( &(cm->messagePipelineList), (void*)mp );

	return 0;
}

tptp_int32 connectionManager_loadTransportLayer( ConnectionManager_t* cm, 
                                          const char*          libName, 
                                          const char*          configData, 
                                          transport_layer_t*   tl )
{
	DLL_REFERENCE           dllRef;
	transport_layer_data_t	tlData;
	char                    fullLibName[_MAX_PATH];

#ifdef _WIN32
	strcpy( fullLibName, libName );
#elif __APPLE__
	sprintf( fullLibName, "lib%s.dylib", libName );
#else
	sprintf( fullLibName, "lib%s.so", libName );
#endif

	dllRef = LOAD_LIBRARY( fullLibName );
	if ( dllRef != NULL )
	{
		/* Initialize our structure */
		tl->dllRef = dllRef;
		tl->transportID = cm->nextTransportID;;
		tl->createTransportListener = (createTransportListener_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "createTransportListener" );
		tl->destroyTransportListener = (destroyTransportListener_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "destroyTransportListener" );
		tl->setProcessMessageFunc = (setProcessMessageFunc_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "setProcessMessageFunc" );
		tl->startTransportListener = (startTransportListener_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "startTransportListener" );
		tl->stopTransportListener = (stopTransportListener_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "stopTransportListener" );
		tl->terminateConnection = (terminateConnection_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "terminateConnection" );
		tl->sendMessage = (sendMessage_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "sendMessage" );
		tl->sendData = (sendData_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "sendData" );
		tl->setIncomingDataFunc = (setIncomingDataFunc_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "setIncomingDataFunc" );
		tl->setDataDescriptor = (setDataDescriptor_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "setDataDescriptor" );
		tl->terminateDataConnection = (terminateDataConnection_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "terminateDataConnection" );
		tl->getPeerConnectionInfo = (getPeerConnectionInfo_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "getPeerConnectionInfo" );
		tl->createPeerConnection = (createPeerConnection_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "createPeerConnection" );
		tl->setPeerAttachInfo = (setPeerAttachInfo_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "setPeerAttachInfo" );

		/* Note that it is OK for the setDataDescriptor,
		   getPeerConnectionInfo, createPeerConnection or setPeerAttachInfo
		   pointers to be NULL.  These are not used for all TLs */

		if ( ( 0 == tl->createTransportListener  ) ||
		     ( 0 == tl->destroyTransportListener ) ||
		     ( 0 == tl->setProcessMessageFunc    ) ||
		     ( 0 == tl->startTransportListener   ) ||
		     ( 0 == tl->stopTransportListener    ) ||
		     ( 0 == tl->terminateConnection      ) ||
		     ( 0 == tl->sendMessage              ) ||
		     ( 0 == tl->sendData                 ) ||
		     ( 0 == tl->setIncomingDataFunc      ) ||
		     ( 0 == tl->terminateDataConnection  ) )
		{
			/* Log an error */
			TPTP_LOG_ERROR_MSG1( cm, "One of more function was missing in transport layer: %s", libName );
			return -1; /* TODO: return a more specific error code */
		}

		/* Build the structure to initialize the TL */
		tlData.transportID = cm->nextTransportID;
		tlData.configurationData = configData;
		tlData.addConnectionEntry = connectionManager_addConnection;
		tlData.removeConnectionEntry = connectionManager_removeConnection;
		tlData.addDataConnectionEntry = connectionManager_addDataConnection;
		tlData.addDirectDataConnectionEntry = connectionManager_addDirectDataConnection;
		tlData.removeDataConnectionEntry = connectionManager_removeDataConnection;
		tlData.logEventEntry = connectionManager_logEvent;

		/* create the transport listener */
		tl->createTransportListener( &(cm->cmo), &tlData, &(tl->tlo) );

		/* Only increment the ID if we were successful */
		cm->nextTransportID++;

		TPTP_LOG_DEBUG_MSG1( cm, "Successfully loaded transport layer: %s", libName );

		return 0;
	}
	else
	{
		TPTP_LOG_ERROR_MSG2( cm, "Unable to load transport layer: %s %s", libName, fullLibName );
	}

	return -1;
}

tptp_int32 connectionManager_loadCommandExtractor( ConnectionManager_t* cm, 
                                            const char*          libName, 
                                            command_extractor_t* cx )
{
	DLL_REFERENCE             dllRef;
	command_extractor_data_t  cxData;
	char                      fullLibName[256];

#ifdef _WIN32
	strcpy( fullLibName, libName );
#elif __APPLE__
	sprintf( fullLibName, "lib%s.dylib", libName );
#else
	sprintf( fullLibName, "lib%s.so", libName );
#endif

	dllRef = LOAD_LIBRARY( fullLibName );
	if ( dllRef != NULL )
	{
		/* Initialize our structure */
		cx->dllRef = dllRef;
		cx->createCommandExtractor = (createCommandExtractor_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "createCommandExtractor" );
		cx->destroyCommandExtractor = (destroyCommandExtractor_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "destroyCommandExtractor" );
		cx->processMessage = (processMessage_ptr_t)RESOLVE_ENTRY_POINT( dllRef, "processMessage" );

		if ( ( 0 == cx->createCommandExtractor  ) ||
		     ( 0 == cx->destroyCommandExtractor ) ||
		     ( 0 == cx->processMessage          ) )
		{
			/* Log an error */
			TPTP_LOG_ERROR_MSG1( cm, "One of more function was missing in command extractor: %s", libName );
			return -1; /* TODO: Return a more specific error code */
		}

		/* Build the structure to initialize the TL */
		cxData.processCommandEntry = (processCommand_ptr_t) connectionManager_processCommand;
		cxData.logEventEntry = connectionManager_logEvent;

		/* create the transport listener */
		cx->createCommandExtractor( &(cm->cmo), &cxData, &(cx->cxo) );

		TPTP_LOG_DEBUG_MSG1( cm, "Successfully loaded command extractor: %s", libName );

		return 0;
	}
	else
	{
		TPTP_LOG_ERROR_MSG2( cm, "Unable to load command extractor: %s %s", libName, fullLibName );
	}

	return -1;
}


tptp_int32 connectionManager_startServers( ConnectionManager_t* cm )
{
	message_pipeline_t* mp;
	tptp_node_t*        node;
	tptp_int32                 ret = 0;

	for ( node = cm->messagePipelineList.head; node != 0; node = node->next )
	{
		mp = (message_pipeline_t*)node->data;

		ret = mp->tl.startTransportListener( &(mp->tl.tlo) );

		if ( ret != 0 )
		{
			/* Log an error */
			TPTP_LOG_ERROR_MSG2( cm, "An error was returned from TransportLayer(%d)::startTransportLayer errNum = %d", mp->tl.transportID, ret );
			return -1;
		}
	}

	return 0;
}

tptp_int32 connectionManager_stopServers( ConnectionManager_t* cm )
{
	message_pipeline_t* mp;
	tptp_node_t*        node;
	tptp_int32          ret = 0;

	for ( node = cm->messagePipelineList.head; node != 0; node = node->next )
	{
		mp = (message_pipeline_t*)node->data;

		ret = mp->tl.stopTransportListener( &(mp->tl.tlo) );

		if ( ret != 0 )
		{
			/* Log an error */
			TPTP_LOG_WARNING_MSG2( cm, "An error was returned from TransportLayer(%d)::stopTransportLayer errNum = %d", mp->tl.transportID, ret );
		}
	}

	return 0;
}

tptp_int32 connectionManager_addConnection( tptp_object* cmo, tptp_uint32 transportID, tptp_uint32* connectionID )
{
	tptp_int32                  id;
	transport_layer_t*   tl = 0;
	connection_t*        connection;
	ConnectionManager_t* cm = (ConnectionManager_t*)cmo->data;

	/* Initialize the connection ID */
	id = cm->nextConnectionID;

	/* Find the transport layer */
	tl = connectionManager_findTransportLayer( cm, transportID );
	if ( tl == 0 )
	{
		/* Log an error */
		TPTP_LOG_ERROR_MSG( cm, "The requested transport ID could not be found during addConnection" );
		return -1; /* TODO: return a more specific error code */
	}

	/* Add a new connection to the list */
	connection = tptp_malloc( sizeof(connection_t) );
	if ( connection == 0 )
	{
		/* Log an error */
		TPTP_LOG_ERROR_MSG( cm, "Out of memory during addConneciton" );
		return -1; /* TODO: return a more specific error code */
	}
	tptp_list_add( &(cm->connectionList), (void *)connection );

	/* Initialize the connection data */
	connection->connectionID = id;
	connection->transport = tl;

	/* Return the new connection ID */
	*connectionID = id;
	TPTP_LOG_DEBUG_MSG2(cm, "ConnectionManger - successfully added connectionID %d for transportID %d",*connectionID, transportID);

	/* Increment the next ID -- prototype */
	cm->nextConnectionID++;

	return 0;
}

tptp_int32 connectionManager_removeConnection( tptp_object* cmo, tptp_uint32 connectionID )
{
	connection_t*        conn;
	ConnectionManager_t* cm = (ConnectionManager_t*)cmo->data;
	
	conn = connectionManager_findConnection( cm, connectionID );
	if ( conn == 0 )
		return -1;
	tptp_list_remove( &(cm->connectionList), (void*)conn );

	return 0;
}

tptp_int32 connectionManager_processCommand( tptp_object* cmo, tptp_uint32 size, const tptp_string* cmd )
{
	tptp_int32                  dest;
	tptp_int32                  ret;
	ConnectionManager_t* cm = (ConnectionManager_t*)cmo->data;

	/* Call a utility function to get the destination ID */
	ret = getDestinationID( cmd, &dest );

	if ( dest == AGENT_MANAGER ) 
	{
		/* Pass this command on to the Agent Manager subcomponent */
		return agentManager_processCommand( cm->agentManager, cmd );
	}
	else if ( dest == LOGGING_SERVICE ) 
	{
		/* Pass this command on to the Logging Service subcomponent */
		return loggingService_processCommand( cm->loggingService, cmd );
	}
	else
	{
		/* Look up the destination and forward it */
		return connectionManager_sendMessage( cm, dest, cmd );
	}
}

/**
 *********************************************************************
 *                                                       
 * @brief                                                
 *    Adds a new entry in the data conneciton table
 *
 * @param transportID   ID of the transport layer that owns the connection
 * @param flags         Bitmask of flags indicate which direction data can go
 * @param connectionID  Pointer to receive the ID assigned to the connection
 *
 * @return
 *    0 if success
 *    nonzero if error
 *
 *********************************************************************/
tptp_int32 connectionManager_addDataConnection( tptp_object* cmo,
                                                tptp_uint32  transportID,
                                                tptp_uint32  flags,
                                                tptp_uint32* connectionID )
{
	ConnectionManager_t* cm = (ConnectionManager_t*)cmo->data;

	if ( ((flags & TPTP_DATA_PATH_DIRECT_SOCKET) || (flags & TPTP_DATA_PATH_DIRECT_FILE)) &&
		 (flags & TPTP_DATA_PATH_RECEIVE) )
	{
		/* This is an error condition.  Use add direct data connection instead. */
		TPTP_LOG_ERROR_MSG( cm, "A non-direct data connection was requested but a direct connection flag was set." );
		return -1;
	}

	/* Pass this on the the more general data connection function */
	connectionManager_addDirectDataConnection( cmo, transportID, flags, 0, connectionID );

	return 0;
}

/**
 *********************************************************************
 *                                                       
 * @brief                                                
 *    Adds a new entry in the data conneciton table
 *
 * @param transportID   ID of the transport layer that owns the connection
 * @param flags         Bitmask of flags indicate which direction data can go
 * @param connectionID  Pointer to receive the ID assigned to the connection
 * @param descriptor    A descriptor that is used if the connection involves 
 *                      direct transfer of data
 *
 * @return
 *    0 if success
 *    nonzero if error
 *
 *********************************************************************/
tptp_int32 connectionManager_addDirectDataConnection( tptp_object* cmo,
                                                      tptp_uint32  transportID,
                                                      tptp_uint32  flags,
                                                      tptp_uint32  descriptor,
                                                      tptp_uint32* connectionID )
{
	tptp_uint32          id;
	transport_layer_t*   tl = 0;
	data_connection_t*   connection;
	ConnectionManager_t* cm = (ConnectionManager_t*)cmo->data;

	if ( ( flags & TPTP_DATA_PATH_DIRECT_SOCKET ) && ( flags & TPTP_DATA_PATH_DIRECT_FILE ) )
	{
		TPTP_LOG_ERROR_MSG( cm, "Conflicting flags were set in a direct data connection request." );
		return -1;
	}

	if ( ( ( descriptor == 0 ) && (flags & TPTP_DATA_PATH_RECEIVE) ) &&
		 ( ( flags & TPTP_DATA_PATH_DIRECT_SOCKET ) || 
	       ( flags & TPTP_DATA_PATH_DIRECT_FILE   ) ) )
	{
		/* 
		   This is an error condition because a direct connection is
		   requested but no corresponding descriptor is provided.
		   The caller should have called addDirectDataConnection with
		   a valid socket descriptor.
		*/
		TPTP_LOG_ERROR_MSG( cm, "A direct data connection was requested but no descriptor was provided" );
		return -1;
	}

	/* Initialize the data connection ID */
	id = cm->nextDataConnectionID;

	/* Find the transport layer */
	tl = connectionManager_findTransportLayer( cm, transportID );
	if ( tl == 0 )
	{
		/* Log an error */
		TPTP_LOG_ERROR_MSG( cm, "The requested transport ID could not be found during addDataConnection" );
		return -1; /* TODO: return a more specific error code */
	}

	/* Add a new connection to the list */
	connection = tptp_malloc( sizeof(data_connection_t) );
	if ( connection == 0 )
	{
		/* Log an error */
		TPTP_LOG_ERROR_MSG( cm, "Out of memory during addDataConnection" );
		return -1; /* TODO: return a more specific error code */
	}
	tptp_list_add( &(cm->dataConnectionList), (void *)connection );

	/* Initialize the connection data */
	connection->connectionID = id;
	connection->transport = tl;
	connection->flags = flags;
	connection->bound = 0;
	connection->descriptor = descriptor;

	/* Return the new connection ID */
	*connectionID = id;

	/* Increment the next ID -- prototype */
	cm->nextDataConnectionID++;

	return 0;
}

tptp_int32 connectionManager_removeDataConnection( tptp_object* cmo, tptp_uint32 connectionID )
{
	ConnectionManager_t* cm = (ConnectionManager_t*)cmo->data;
	data_connection_t*   conn;
	
	conn = connectionManager_findDataConnection( cm, connectionID );
	if ( conn == 0 )
		return -1;
	tptp_list_remove( &(cm->dataConnectionList), (void*)conn );

	return 0;
}

/**
 *********************************************************************
 *                                                       
 * @brief                                                
 *    This method is called by internal components of the Agent
 *    Controller to send messages to external components
 *
 * @param connectionID  The ID of the connection to which to send the message
 * @param msg           The message to be sent, as an XML fragment
 *
 * @return
 *    0 if success
 *    nonzero if error
 *
 *********************************************************************/

tptp_int32 connectionManager_sendMessage( ConnectionManager_t* cm, tptp_uint32 connectionID, const tptp_string* msg )
{
	connection_t* conn;

	conn = connectionManager_findConnection( cm, connectionID );
	if ( conn == 0 )
	{
		/* Log an error */
		TPTP_LOG_ERROR_MSG1( cm, "The requested connection (%ud) could not be found during sendMessage", connectionID );
		return -1; /* TODO: return a more specific error code */
	}

	(conn->transport->sendMessage)( &(conn->transport->tlo), connectionID, strlen(msg), msg );

	return 0;
}

/**
 *********************************************************************
 *                                                       
 * @brief                                                
 *    Bind two data connections together, directing the 
 *    output from one to the input of the other, and vice
 *    versa
 *
 * @param sourceID   The ID of the object initiating the bind
 * @param context    The context ID for the incoming message
 * @param dataConn1  The ID of the first data connection to bind
 * @param dataConn2  The ID of the second data connection to bind
 *
 * @return
 *    0 if success
 *    nonzero if error
 *
 *********************************************************************/

tptp_int32 connectionManager_bindDataConnections( ConnectionManager_t* cm,  
                                                  tptp_uint32          sourceID, 
                                                  tptp_uint32          context, 
                                                  tptp_uint32          dataConn1, 
                                                  tptp_uint32          dataConn2 )
{
	char     responseFormat[] = "<Cmd src=\"%ld\" dest=\"%ld\" ctxt=\"%ld\"><dataConnectionsBound iid=\"org.eclipse.tptp.agentManager\"></dataConnectionsBound></Cmd>";
	char     responseCmd[256];
	data_connection_t* conn1 = 0;
	data_connection_t* conn2 = 0;

	TPTP_LOG_DEBUG_MSG( cm, "Received bindDataConnection request" );

	conn1 = connectionManager_findDataConnection( cm, dataConn1 );
	conn2 = connectionManager_findDataConnection( cm, dataConn2 );

	if ( (conn1 == 0) || (conn2 == 0) )
	{
		/* Report this to the caller */
		connectionManager_sendErrorResponse( cm, TPTP_AM_DATA_CONNECTION_NOT_FOUND, sourceID, context );
		return -1;
	}


	if ( ( (conn1->flags & TPTP_DATA_PATH_SEND)          && !(conn2->flags & TPTP_DATA_PATH_RECEIVE)       ) ||
	     ( (conn1->flags & TPTP_DATA_PATH_RECEIVE)       && !(conn2->flags & TPTP_DATA_PATH_SEND)          ) ||
	     ( (conn2->flags & TPTP_DATA_PATH_SEND)          && !(conn1->flags & TPTP_DATA_PATH_RECEIVE)       ) ||
	     ( (conn2->flags & TPTP_DATA_PATH_RECEIVE)       && !(conn1->flags & TPTP_DATA_PATH_SEND)          ) ||
		 ( (conn2->flags & TPTP_DATA_PATH_MULTIPLEXED)   && !(conn1->flags & TPTP_DATA_PATH_MULTIPLEXED)   ) ||
		 ( (conn2->flags & TPTP_DATA_PATH_DIRECT_FILE)   && !(conn1->flags & TPTP_DATA_PATH_DIRECT_FILE)   ) ||
		 ( (conn2->flags & TPTP_DATA_PATH_DIRECT_SOCKET) && !(conn1->flags & TPTP_DATA_PATH_DIRECT_SOCKET) ) )
	{
		/* Report this to the caller */
		connectionManager_sendErrorResponse( cm, TPTP_AM_DATA_CONNECTIONS_INCOMPATIBLE, sourceID, context );
		return -1;
	}

	if ( !(conn1->flags & TPTP_DATA_PATH_RECEIVE) && !(conn2->flags & TPTP_DATA_PATH_RECEIVE) )
	{
		/* Report this to the caller */
		connectionManager_sendErrorResponse( cm, TPTP_AM_NO_DATA_RECEIVER, sourceID, context );
		return -1;
	}

	/* Hook up the data path */
	if ( (conn1->flags & TPTP_DATA_PATH_DIRECT_FILE) || (conn1->flags & TPTP_DATA_PATH_DIRECT_SOCKET) )
	{
		if ( !conn2->flags & TPTP_DATA_PATH_RECEIVE )
		{
			/* Report this to the caller */
			connectionManager_sendErrorResponse( cm, TPTP_AM_DATA_CONNECTIONS_INCOMPATIBLE, sourceID, context );
			return -1;
		}

		if ( conn2->transport->setDataDescriptor == NULL )
		{
			/* Report this to the caller */
			connectionManager_sendErrorResponse( cm, TPTP_AM_TL_MISSING_FUNC, sourceID, context );
			return -1;
		}

		if ( conn1->descriptor == 0 )
		{
			/* Report this to the caller */
			connectionManager_sendErrorResponse( cm, TPTP_AM_TL_MISSING_DESCRIPTOR, sourceID, context );
			return -1;
		}

		conn2->transport->setDataDescriptor( &(conn2->transport->tlo), dataConn2, dataConn1, &(conn1->transport->tlo), conn1->descriptor );
	}
	else
	{
		if ( conn1->flags & TPTP_DATA_PATH_RECEIVE )
		{
			conn2->transport->setIncomingDataFunc( &(conn2->transport->tlo), dataConn2, dataConn1, &(conn1->transport->tlo), conn1->transport->sendData );
		}
		if ( conn2->flags & TPTP_DATA_PATH_RECEIVE )
		{
			conn1->transport->setIncomingDataFunc( &(conn1->transport->tlo), dataConn1, dataConn2, &(conn2->transport->tlo), conn2->transport->sendData );
		}
	}

	/* Send the response */
	sprintf( responseCmd, responseFormat, AGENT_MANAGER, sourceID, context );
	connectionManager_sendMessage( cm, sourceID, responseCmd );

	return 0;
}

transport_layer_t* connectionManager_findTransportLayer( ConnectionManager_t* cm, tptp_uint32 id )
{
	message_pipeline_t* mp;
	tptp_node_t*        node;

	for ( node = cm->messagePipelineList.head; node != 0; node = node->next )
	{
		mp = (message_pipeline_t*)node->data;
		if ( mp->tl.transportID == id )
		{
			return &(mp->tl);
		}
	}

	return 0;
}


connection_t* connectionManager_findConnection( ConnectionManager_t* cm, tptp_uint32 id )
{
	connection_t* conn;
	tptp_node_t*  node;

	for (node = cm->connectionList.head; node != 0; node = node->next )
	{
		conn = (connection_t*)node->data;
		if ( conn->connectionID == id )
		{
			return conn;
		}
	}

	return 0;
}

data_connection_t* connectionManager_findDataConnection( ConnectionManager_t* cm, tptp_uint32 id )
{
	data_connection_t* conn;
	tptp_node_t*       node;

	for (node = cm->dataConnectionList.head; node != 0; node = node->next )
	{
		conn = (data_connection_t*)node->data;
		if ( conn->connectionID == id )
		{
			return conn;
		}
	}

	return 0;
}

tptp_int32 connectionManager_logEvent( tptp_object* cmo, 
                                       tptp_string* subcomponent,
                                       tptp_uint32  instanceId,
                                       tptp_string* filename,
                                       tptp_int32   lineNum,
                                       tptp_int32   severity, 
                                       tptp_string* event )
{
	ConnectionManager_t* cm = (ConnectionManager_t*)cmo->data;

	// The TPTP_AC_LOG macro isn't used here because it takes the current 
	//    file and line number and we need to use those passed in here
	loggingService_logEvent( cm->loggingService, subcomponent, instanceId, filename, lineNum, severity, event );

	return 0;
}

void connectionManager_sendErrorResponse( ConnectionManager_t* cm, tptp_int32 errorCode, tptp_int32 sourceID, tptp_int32 contextID )
{
	tptp_string* cbeStr;
	tptp_int32   rc;
	char*        errorMsg;

	errorMsg = tptp_getErrorString( errorCode );

	TPTP_LOG_ERROR_MSG( cm, errorMsg );

	rc = tptp_createErrorCmdCBE( "Agent Controller", "Connection Manager", CONNECTION_MANAGER, TPTP_CRITICAL, errorMsg, &cbeStr );

	if ( rc == 0 )
	{
		char errorFormat[] = "<Cmd src=\"%d\" dest=\"%d\" ctxt=\"%d\"><TPTP_Error iid=\"org.eclipse.tptp.Error\"><ErrCode>%d</ErrCode><ErrInfo>%s</ErrInfo></TPTP_Error></Cmd>";
		char errorCmd[8192];

		/* Ask this agent to go away */
		sprintf( errorCmd, errorFormat, AGENT_MANAGER, sourceID, contextID, errorCode, cbeStr );
		connectionManager_sendMessage( cm, sourceID, errorCmd );
	}
	else
	{
		/* If we couldn't build the CBE string, send the error with a raw error message */
		char errorFormat[] = "<Cmd src=\"%d\" dest=\"%d\" ctxt=\"%d\"><TPTP_Error iid=\"org.eclipse.tptp.Error\"><ErrCode>%d</ErrCode><ErrInfo>%s</ErrInfo></TPTP_Error></Cmd>";
		char errorCmd[8192];

		/* Ask this agent to go away */
		sprintf( errorCmd, errorFormat, AGENT_MANAGER, sourceID, contextID, errorCode, errorMsg );
		connectionManager_sendMessage( cm, sourceID, errorCmd );
	}

	tptp_free( errorMsg );
}

tptp_int32 connectionManager_initializePeerConnectionTransport( ConnectionManager_t* cm, char* peerConnectionTransport )
{
	transport_layer_t* tl;

	if ( peerConnectionTransport == NULL )
	{
		TPTP_LOG_ERROR_MSG( cm, "A NULL argument was passed to connectionManager_initializePeerConnectionTransport." );
		return TPTP_UNEXPECTED_NULL_ARG;
	}

	tl = connectionManager_findTransportLayerByType( cm, peerConnectionTransport );
	if ( tl == NULL )
	{
		TPTP_LOG_ERROR_MSG1( cm, "Unable to locate transport type, %s, in connectionManager_initializePeerConnectionTransport.", peerConnectionTransport );
		return -1;
	}

	if ( !tl->setPeerAttachInfo )
	{
		TPTP_LOG_ERROR_MSG1( cm, "The transport layer specified for peer connection, %s, does not provide the setPeerAttachInfo function.", peerConnectionTransport );
		return -1;
	}

	return tl->setPeerAttachInfo( &tl->tlo, &cm->peerInfo );
}

tptp_int32 connectionManager_getPeerConnectionInfo( ConnectionManager_t* cm, 
                                                    tptp_string*         pmt, 
                                                    tptp_string*         connectionType,
                                                    tptp_string**        ci )
{
	transport_layer_t* tl;

	if ( ( ci == NULL ) || (connectionType == NULL) || ( pmt == NULL ) )
	{
		TPTP_LOG_ERROR_MSG( cm, "A NULL argument was passed to connectionManager_getPeerConnectionInfo." );
		return TPTP_UNEXPECTED_NULL_ARG;
	}

	tl = connectionManager_findTransportLayerByType( cm, pmt );
	if ( tl == NULL )
	{
		TPTP_LOG_ERROR_MSG1( cm, "Unable to locate transport type, %s, in connectionManager_getPeerConnectionInfo.", pmt );
		return -1;
	}

	if ( !tl->getPeerConnectionInfo )
	{
		TPTP_LOG_ERROR_MSG1( cm, "The transport layer specified for peer connection, %s, does not provide the getPeerConnectionInfo function.", pmt );
		return -1;
	}

	return tl->getPeerConnectionInfo( &tl->tlo, connectionType, ci );
}

tptp_int32 connectionManager_connectToPeer( ConnectionManager_t* cm, tptp_string* sci, tptp_uint32* connectionID  )
{
	transport_layer_t* tl;
	tptp_string*       pmt;

	if ( 0 != getXmlFragmentElementString( sci, "sourceConnectionInfo", "transportType", &pmt ) )
	{
		/* The connection might be requested for shutdown */
		if ( 0 != getXmlFragmentElementString( sci, SHUTDOWNCONNECTION_XML_NAME, "transportType", &pmt ) )
		{
			TPTP_LOG_ERROR_MSG( cm, "The peerConnectionInfo provided to connectionManager_connectToPeer does not contain transportType." );
			return -1;
		}
	}

	tl = connectionManager_findTransportLayerByType( cm, pmt );
	tptp_free( pmt );
	if ( tl == NULL )
	{
		TPTP_LOG_ERROR_MSG1( cm, "Unable to locate transport type, %s, in connectionManager_connectToPeer.", pmt );
		return -1;
	}

	if ( !tl->createPeerConnection )
	{
		TPTP_LOG_ERROR_MSG1( cm, "The transport layer specified for peer monitoring, %s, does not provide the createPeerConnection function.", pmt );
		return -1;
	}

	return tl->createPeerConnection( &tl->tlo, sci, connectionID );
}

tptp_int32 connectionManager_dropPeerConnection( ConnectionManager_t* cm, tptp_uint32 connectionID )
{
	connection_t* connection;
	
	connection = connectionManager_findConnection( cm, connectionID );
	if ( connection == NULL )
	{
		TPTP_LOG_ERROR_MSG1( cm, "The requested connection (%ud) could not be found during dropPeerConnection", connectionID );
		return -1;
	}

	return connection->transport->terminateConnection( &(connection->transport->tlo), connectionID );
}

transport_layer_t* connectionManager_findTransportLayerByType( ConnectionManager_t* cm, tptp_string* type )
{
	tptp_node_t* node;

	if ( type == NULL )
		return NULL;

	for ( node = cm->messagePipelineList.head; node != NULL; node = node->next )
	{
		message_pipeline_t* mp = (message_pipeline_t*)node->data;

		if ( (mp->tl.type != NULL) && isEqualString(mp->tl.type, type) )
		{
			return &(mp->tl);
		}
	}

	return NULL;
}
