forwarderPort.h

Go to the documentation of this file.

/*******************************************************************************
** Copyright (c) 2005 MaK Technologies, Inc.
** All rights reserved.
*******************************************************************************/
/*******************************************************************************
** $RCSfile: forwarderPort.h,v $ $Revision: 1.18 $ $State: Exp $
*******************************************************************************/

#ifndef DtForwarderPort_H_
#define DtForwarderPort_H_

#include "routeMap.h"
#include "rtiMsConfig.h"
#include "rtiVersion.h"

#include <vlutil/vlConfig.h>
#include <vlutil/vlUtil.h>
#include <vlutil/vlSelectParamW.h>
#include <vlutil/vlFedTarget.h>
#include <vlutil/vlInetAddr.h>
#include <map>
#include <vector>
#include <algorithm>

class DtRIDParameters;
class DtRouteMap;
class DtTcpMsgSocket;

class DT_DLL_FORWARDER DtForwarderPort
{
public:
   
   typedef DtForwarderPort* (*DtForwarderPortCreatorFcn)( const DtRIDParameters*,
      const DtBoost::shared_ptr<DtRouteMap>, 
      const MAKRti::DtInetAddr&, int, const MAKRti::DtInetAddr& );

   struct DtForwarderMsg
   {
      caddr_t msg;
      int length;
      MAKRti::DtTransportType transport;
   };

   typedef DtRouteMap::DtAddressPortList           DtAddressPortList;
   typedef DtRouteMap::DtAddressList               DtAddressList;
   typedef DtRouteMap::DtInterfaceAddrSet          DtInterfaceAddrSet;
   typedef std::vector<DtTcpMsgSocket*>            DtSocketList;
   typedef std::set<DtTcpMsgSocket*>               DtSocketSet;
   typedef std::pair< DtInetAddrAndPort, DtForwarderMsg > DtAddressMsgPair;
   typedef std::pair<DtTcpMsgSocket*, DtForwarderMsg> DtSockMsgPair;     
   typedef std::map< DtInetAddrAndPort, DtTcpMsgSocket* > DtAddrToConnMap;
   typedef std::map<DtInetAddrAndPort, MAKRti::DtString> DtConnAndNameMap;

   typedef int (*DtConnectionChangeCb)(const MAKRti::DtString&, unsigned int,
      const DtInetAddrAndPort&, void*);   
   typedef void (*DtFederateAddedCb)(const MAKRti::DtString&,
      const MAKRti::DtString&, const MAKRti::DtString&, unsigned int,
      unsigned int, const DtInetAddrAndPort&, const MAKRti::DtString&,
      unsigned int, const MAKRti::DtString&, const MAKRti::DtString&,
      DtHlaVersionId, void*);
   typedef void (*DtFederateRemovedCb)(const MAKRti::DtString&,
      const MAKRti::DtString&, unsigned int, unsigned int, 
      const DtInetAddrAndPort&, void*);
   typedef void (*DtPortReadyCb)(unsigned long, void*);
   typedef int (*DtExecAddChangeCb)(const MAKRti::DtString&, unsigned int,
      const DtInetAddrAndPort&, const MAKRti::DtString&, bool, void*);
   typedef int (*DtExecRemovedChangeCb)(const MAKRti::DtString&, unsigned int,
      const DtInetAddrAndPort&, void*);
   typedef int (*DtSocketQueueCb)(unsigned int, int, void*);
   typedef void (*DtWaitingForFwsCb)(const DtConnAndNameMap&,
      const DtConnAndNameMap&, void*);
   typedef void (*DtNotWaitingForFwsCb)(void*);
   typedef std::vector< std::pair< DtConnectionChangeCb, void* > > DtCallbackList;
   typedef std::vector< std::pair< DtFederateAddedCb, void* > > DtFederateAddedCallbackList;
   typedef std::vector< std::pair< DtFederateRemovedCb, void* > > DtFederateRemovedCallbackList;
   typedef std::vector< std::pair< DtPortReadyCb, void*> > DtPortReadyCallbackList;
   typedef std::vector< std::pair< DtExecAddChangeCb, void* > > DtExecAddedCallbackList;  
   typedef std::vector< std::pair< DtExecRemovedChangeCb, void* > > DtExecRemovedCallbackList;  
   typedef std::vector< std::pair< DtSocketQueueCb, void* > > DtSocketQueueCallbackList;
   typedef std::vector< std::pair< DtWaitingForFwsCb, void* > > DtWaitingForFwsCallbackList;
   typedef std::vector< std::pair< DtNotWaitingForFwsCb, void* > > DtNotWaitingForFwsCallbackList;

protected:  
   DtForwarderPort( const DtRIDParameters* params, DtBoost::shared_ptr<DtRouteMap> routes, 
      MAKRti::DtInetAddr host, int forwarderPort, MAKRti::DtInetAddr addnlConnection ) :
      myParams( params ), myRoutes( routes ), myAddress( host ), myOk( true ), 
         myPortToUse(forwarderPort), myAdditionalAddrToConnectTo(addnlConnection) 
   {}
      
public:
   virtual ~DtForwarderPort() {};
   
   virtual bool isOk() const { return myOk; }
   
   virtual bool isReady() const = 0;

   virtual DtAddressMsgPair getNextMsg() = 0;

   virtual void sendByTransport( const DtForwarderMsg& msg ) const = 0;

   virtual void sendToAllReliable( const DtForwarderMsg& msg ) const = 0;

   virtual void sendToListReliable( const DtForwarderMsg& msg,
      const DtRouteMap::DtAddressPortList& destinations ) const = 0;

   virtual inline bool flush();

   virtual void selectParams( MAKRti::DtSelectParamWithWrite& selectParams ) = 0;
   
   virtual void updateRoutes() = 0;

   virtual int checkConnectionStatus( const DtInetAddrAndPort* target = 0 ) = 0;

   virtual void clearNeededConnections() = 0;

   virtual unsigned int abortPendingConnections() = 0;

   virtual unsigned int pendingConnectionsAborted() = 0;

   virtual DtAddressPortList connections() const = 0;

   virtual void listenToMulticastGroup( const MAKRti::DtInetAddr& mcastAddr ) = 0;
   virtual void dropMulticastGroup( const MAKRti::DtInetAddr& mcastAddr ) = 0;

   virtual bool updateFederateConnectionId( unsigned long orig, 
                                            unsigned long final ) = 0;

   virtual inline void addConnectionEstablishedCb( DtConnectionChangeCb callback, void* usr );
   virtual inline void removeConnectionEstablishedCb( DtConnectionChangeCb callback, void* usr );
   
   virtual inline void addConnectionLostCb( DtConnectionChangeCb callback, void* usr );
   virtual inline void  removeConnectionLostCb( DtConnectionChangeCb callback, void* usr );

   virtual inline void addForwarderPortReadyCb( DtPortReadyCb callback, void* usr );
   virtual inline void removeForwarderPortReadyCb( DtPortReadyCb callback, void* usr );

   virtual inline void addFederateJoinedCb( DtFederateAddedCb callback, void* usr );
   virtual inline void removeFederateJoinedCb( DtFederateAddedCb callback, void* usr );

   virtual inline void addFederateResignedCb( DtFederateRemovedCb callback, void* usr );
   virtual inline void removeFederateResignedCb( DtFederateRemovedCb callback, void* usr );

   virtual inline void addSocketQueueCb( DtSocketQueueCb callback, void* usr );
   virtual inline void removeSocketQueueCb( DtSocketQueueCb callback, void* usr );
   
   virtual inline void addWaitingForForwarderConnectionsCb(DtWaitingForFwsCb callback, void* usr);
   virtual inline void removeWaitingForForwarderConnectionsCb(DtWaitingForFwsCb callback, void* usr);

   virtual inline void addNotWaitingForForwarderConnectionsCb(DtNotWaitingForFwsCb callback, void* usr);
   virtual inline void removeNotWaitingForForwarderConnectionsCb(DtNotWaitingForFwsCb callback, void* usr);

   virtual inline void addExecConnectionEstablishedCb( 
      DtExecAddChangeCb callback, void* usr);
   virtual inline void removeExecConnectionEstablishedCb( 
      DtExecAddChangeCb callback, void * usr);

   virtual inline void addExecDisconnectedCb(
      DtExecRemovedChangeCb callback, void* usr);
   virtual inline void removeExecDisconnectedCb(
      DtExecRemovedChangeCb callback, void* usr);

   virtual int execConnected( const MAKRti::DtString& connName,
      unsigned int connID, const DtInetAddrAndPort& addr, const MAKRti::DtString& hostName,
      bool fullyCompliant);
   virtual int execDisconnected( const MAKRti::DtString& connName,
      unsigned int connID, const DtInetAddrAndPort& addr );

   virtual int socketQueueStatus( unsigned int connId, int connQueueStatus );

   virtual void forwarderNodeIsReady(unsigned long id);

   virtual void waitingForForwarderConnections(const DtConnAndNameMap& allNeededFws,
      const DtConnAndNameMap& stillNeededFws);

   virtual void notWaitingForForwarderConnections();


   virtual void setForwarderId(unsigned long id);

protected:

   virtual inline int connectionEstablished( const MAKRti::DtString& connName,
      unsigned int connID, const DtInetAddrAndPort& addr );
   virtual inline int connectionLost( const MAKRti::DtString& connName,
      unsigned int connID, const DtInetAddrAndPort& addr );

   static int connectionEstablished( const MAKRti::DtString& connName,
      unsigned int connID, const DtInetAddrAndPort& addr, void* usr );
   static int connectionLost( const MAKRti::DtString& connName,
      unsigned int connID, const DtInetAddrAndPort& addr, void* usr );
   
   virtual void federateResigned( const MAKRti::DtString& connName,
      const MAKRti::DtString& connType,unsigned int connID, 
      unsigned int fedHandle, const DtInetAddrAndPort& addr );
   virtual void federateJoined( const MAKRti::DtString& connName,
      const MAKRti::DtString& connType,
      const MAKRti::DtString& federationName, unsigned int connID, 
      unsigned int fedHandle, const DtInetAddrAndPort& addr,
      const MAKRti::DtString& hostName, unsigned int processId,
      const MAKRti::DtString& processName,
      const MAKRti::DtString& commandArgs, DtHlaVersionId hlaVersion );

   static void federateJoined( const MAKRti::DtString& connName,
      const MAKRti::DtString& connType,
      const MAKRti::DtString& federationName, unsigned int connID,
      unsigned int fedHandle, const DtInetAddrAndPort& addr,
      const MAKRti::DtString& hostName, unsigned int processId,
      const MAKRti::DtString& processName, const MAKRti::DtString& commandArgs,
      DtHlaVersionId hlaVersion, void* usr);
   static void federateResigned( const MAKRti::DtString& connName,
      const MAKRti::DtString& connType,unsigned int connID, 
      unsigned int fedHandle, const DtInetAddrAndPort& addr, void* usr);
   
   static int execConnected( const MAKRti::DtString& connName,
      unsigned int connID, const DtInetAddrAndPort& addr,
      const MAKRti::DtString& hostName, bool fullyCompliant, void* usr);
   static int execDisconnected( const MAKRti::DtString& connName,
      unsigned int connID, const DtInetAddrAndPort& addr, void* usr);

   static int socketQueueStatus( unsigned int connId, int connQueueStatus,
      void* usr );

   static void forwarderNodeIsReady( unsigned long id, void* usr);

protected:

   const DtRIDParameters* myParams;
   DtBoost::shared_ptr<DtRouteMap> myRoutes;
   const MAKRti::DtInetAddr        myAddress;
   bool                            myOk;

   int myPortToUse;
   MAKRti::DtInetAddr myAdditionalAddrToConnectTo;

   DtCallbackList           myConnectionEstablishedCbs;
   DtCallbackList           myConnectionLostCbs;
   DtFederateAddedCallbackList   myFederateJoinedCbs;
   DtFederateRemovedCallbackList   myFederateResignedCbs;
   DtPortReadyCallbackList  myPortReadyCbs;
   DtExecAddedCallbackList  myExecConnectedCbs;
   DtExecRemovedCallbackList myExecDisconnectedCbs;
   DtSocketQueueCallbackList mySocketQueueCbs;
   DtWaitingForFwsCallbackList myWaitingForFwsCbs;
   DtNotWaitingForFwsCallbackList myNotWaitingForFwsCbs;
};

inline bool DtForwarderPort::flush()
{
   return false;
}


inline void DtForwarderPort::setForwarderId(unsigned long id)
{
   // Do nothing.
   return;
}

inline void DtForwarderPort::addConnectionEstablishedCb( DtConnectionChangeCb callback, void* usr )
{
   std::pair<DtConnectionChangeCb, void*> cbPair( callback, usr );
   DtCallbackList::iterator pos = std::find( myConnectionEstablishedCbs.begin(),
      myConnectionEstablishedCbs.end(), cbPair);   
   if ( pos == myConnectionEstablishedCbs.end() )
   {
      myConnectionEstablishedCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeConnectionEstablishedCb( DtConnectionChangeCb callback, void* usr )
{
   DtCallbackList::iterator pos = std::find( myConnectionEstablishedCbs.begin(),
      myConnectionEstablishedCbs.end(), std::make_pair( callback, usr ));
   if ( pos != myConnectionEstablishedCbs.end() ) myConnectionEstablishedCbs.erase( pos );
}


inline int DtForwarderPort::connectionEstablished( const MAKRti::DtString& connName,
                                                   unsigned int connID,
                                                   const DtInetAddrAndPort& addr )
{
   int retVal = 0;
   for ( DtCallbackList::iterator iter = myConnectionEstablishedCbs.begin();
      iter != myConnectionEstablishedCbs.end(); ++iter )
   {
      if (-1 == ( *iter->first )( connName, connID, addr, iter->second ))
      {
         retVal = -1;
      }      
   }
   return retVal;
}


inline void DtForwarderPort::addConnectionLostCb( DtConnectionChangeCb callback, void* usr )
{
   std::pair<DtConnectionChangeCb, void*> cbPair( callback, usr );
   DtCallbackList::iterator pos = std::find( myConnectionLostCbs.begin(),
      myConnectionLostCbs.end(), cbPair);   
   if ( pos == myConnectionLostCbs.end() )
   {
      myConnectionLostCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeConnectionLostCb( DtConnectionChangeCb callback, void* usr )
{
   DtCallbackList::iterator pos = std::find( myConnectionLostCbs.begin(),
      myConnectionLostCbs.end(), std::make_pair( callback, usr ));
   if ( pos != myConnectionLostCbs.end() ) myConnectionLostCbs.erase( pos );
}

inline int DtForwarderPort::connectionLost( const MAKRti::DtString& connName,
                                            unsigned int connID,
                                            const DtInetAddrAndPort& addr )
{
   int retVal(0);
   for ( DtCallbackList::iterator iter = myConnectionLostCbs.begin();
         iter != myConnectionLostCbs.end(); ++iter )
   {
      if ( -1 == ( *iter->first )( connName, connID, addr, iter->second ) )
      {
         retVal = -1;
      }
   }
   return retVal;
}



inline void DtForwarderPort::addFederateResignedCb( DtFederateRemovedCb callback, void* usr )
{
   std::pair<DtFederateRemovedCb, void*> cbPair( callback, usr );
   DtFederateRemovedCallbackList::iterator pos = std::find( myFederateResignedCbs.begin(),
      myFederateResignedCbs.end(), cbPair);   
   if ( pos == myFederateResignedCbs.end() )
   {
      myFederateResignedCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeFederateResignedCb( DtFederateRemovedCb callback, void* usr )
{
   DtFederateRemovedCallbackList::iterator pos = std::find( myFederateResignedCbs.begin(),
      myFederateResignedCbs.end(), std::make_pair( callback, usr ));
   if ( pos != myFederateResignedCbs.end() ) myFederateResignedCbs.erase( pos );
}

inline void DtForwarderPort::addSocketQueueCb( DtSocketQueueCb callback, void* usr )
{
    std::pair< DtSocketQueueCb, void* > cbPair( callback, usr );

    DtSocketQueueCallbackList::iterator pos = std::find( mySocketQueueCbs.begin(),
        mySocketQueueCbs.end(), cbPair );

    if( pos == mySocketQueueCbs.end() )
        mySocketQueueCbs.push_back( cbPair );
}

inline void DtForwarderPort::removeSocketQueueCb( DtSocketQueueCb callback, void* usr )
{
    DtSocketQueueCallbackList::iterator pos = std::find( mySocketQueueCbs.begin(),
        mySocketQueueCbs.end(), std::make_pair( callback, usr ));

    if( pos != mySocketQueueCbs.end() ) 
        mySocketQueueCbs.erase( pos );
}

inline void DtForwarderPort::federateResigned( const MAKRti::DtString& connName, const MAKRti::DtString& connType, unsigned int connID, 
                                              unsigned int fedHandle, const DtInetAddrAndPort& addr )
{
   for ( DtFederateRemovedCallbackList::iterator iter = myFederateResignedCbs.begin();
      iter != myFederateResignedCbs.end(); ++iter )
   {
      ( *iter->first )( connName, connType, connID, fedHandle, addr, iter->second );
   }
}

inline void DtForwarderPort::federateResigned(const MAKRti::DtString& connName, const MAKRti::DtString& connType, unsigned int connID, 
                           unsigned int fedHandle, const DtInetAddrAndPort& addr, void* usr)
{
   ((DtForwarderPort*)usr)->federateResigned(connName, connType, connID, fedHandle, addr);
}

inline void DtForwarderPort::addForwarderPortReadyCb( DtPortReadyCb callback, void* usr )
{
   std::pair<DtPortReadyCb, void*> cbPair( callback, usr );
   DtPortReadyCallbackList::iterator pos = std::find( myPortReadyCbs.begin(),
      myPortReadyCbs.end(), cbPair);   
   if ( pos == myPortReadyCbs.end() )
   {
      myPortReadyCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeForwarderPortReadyCb( DtPortReadyCb callback, void* usr )
{
   DtPortReadyCallbackList::iterator pos = std::find( myPortReadyCbs.begin(),
      myPortReadyCbs.end(), std::make_pair( callback, usr ));
   if ( pos != myPortReadyCbs.end() ) myPortReadyCbs.erase( pos );
}


inline void DtForwarderPort::forwarderNodeIsReady(unsigned long id, void* usr)
{
   return (static_cast<DtForwarderPort*>(usr))->forwarderNodeIsReady(id);
}

inline void DtForwarderPort::forwarderNodeIsReady(unsigned long id)
{
   for ( DtPortReadyCallbackList::iterator iter = myPortReadyCbs.begin();
      iter != myPortReadyCbs.end(); ++iter )
   {
      ( *iter->first )( id, iter->second);
   }
}



inline void DtForwarderPort::addFederateJoinedCb( DtFederateAddedCb callback, void* usr )
{
   std::pair<DtFederateAddedCb, void*> cbPair( callback, usr );
   DtFederateAddedCallbackList::iterator pos = std::find( myFederateJoinedCbs.begin(),
      myFederateJoinedCbs.end(), cbPair);   
   if ( pos == myFederateJoinedCbs.end() )
   {
      myFederateJoinedCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeFederateJoinedCb( DtFederateAddedCb callback, void* usr )
{
   DtFederateAddedCallbackList::iterator pos = std::find( myFederateJoinedCbs.begin(),
      myFederateJoinedCbs.end(), std::make_pair( callback, usr ));
   if ( pos != myFederateJoinedCbs.end() ) myFederateJoinedCbs.erase( pos );
}

inline void DtForwarderPort::federateJoined(const MAKRti::DtString& connName,
                                            const MAKRti::DtString& connType,
                                            const MAKRti::DtString& federationName,
                                            unsigned int connID, 
                                            unsigned int fedHandle,
                                            const DtInetAddrAndPort& addr,
                                            const MAKRti::DtString& hostName,
                                            unsigned int processId,
                                            const MAKRti::DtString& processName,
                                            const MAKRti::DtString& commandArgs,
                                            DtHlaVersionId hlaVersion )
{
   for ( DtFederateAddedCallbackList::iterator iter = myFederateJoinedCbs.begin();
      iter != myFederateJoinedCbs.end(); ++iter )
   {
      ( *iter->first )( connName, connType, federationName, connID, fedHandle,
         addr, hostName, processId, processName, commandArgs, hlaVersion,
         iter->second );
   }
}

inline void DtForwarderPort::federateJoined(const MAKRti::DtString& connName,
                                            const MAKRti::DtString& connType,
                                            const MAKRti::DtString& federationName,
                                            unsigned int connID,
                                            unsigned int fedHandle,
                                            const DtInetAddrAndPort& addr,
                                            const MAKRti::DtString& hostName, 
                                            unsigned int processId,
                                            const MAKRti::DtString& processName,
                                            const MAKRti::DtString& commandArgs,
                                            DtHlaVersionId hlaVersion,
                                            void* usr)
{
   ((DtForwarderPort*)usr)->federateJoined(connName, connType, federationName,
      connID, fedHandle, addr, hostName, processId, processName, commandArgs,
      hlaVersion);
}


inline int DtForwarderPort::connectionEstablished( const MAKRti::DtString& connName,
                                                   unsigned int connID,
                                                   const DtInetAddrAndPort& addr,
                                                   void* usr )
{
   return ((DtForwarderPort*)usr)->connectionEstablished(connName, connID, addr);
}

inline int DtForwarderPort::connectionLost( const MAKRti::DtString& connName,
                                            unsigned int connID,
                                            const DtInetAddrAndPort& addr,
                                            void* usr )
{
   return ((DtForwarderPort*)usr)->connectionLost(connName, connID, addr);
}



inline void DtForwarderPort::addExecConnectionEstablishedCb( DtExecAddChangeCb callback, void* usr )
{
   std::pair<DtExecAddChangeCb, void*> cbPair( callback, usr );
   DtExecAddedCallbackList::iterator pos = std::find( myExecConnectedCbs.begin(),
      myExecConnectedCbs.end(), cbPair);   
   if ( pos == myExecConnectedCbs.end() )
   {
      myExecConnectedCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeExecConnectionEstablishedCb( DtExecAddChangeCb callback, void* usr )
{
   DtExecAddedCallbackList::iterator pos = std::find( myExecConnectedCbs.begin(),
      myExecConnectedCbs.end(), std::make_pair( callback, usr ));
   if ( pos != myExecConnectedCbs.end() )
   {
      myExecConnectedCbs.erase( pos );
   }
}

inline int DtForwarderPort::execConnected( const MAKRti::DtString& connName, 
                                           unsigned int connID, 
                                           const DtInetAddrAndPort& addr, 
                                           const MAKRti::DtString& hostName,
                                           bool fullyCompliant, void* usr)
{
   return ((DtForwarderPort*)usr)->execConnected(connName, connID, addr, hostName, fullyCompliant);
}

inline int DtForwarderPort::execConnected( const MAKRti::DtString& connName, 
                                           unsigned int connID,
                                           const DtInetAddrAndPort& addr,
                                           const MAKRti::DtString& hostName,
                                           bool fullyCompliant )
{
   int retVal = 0;
   for ( DtExecAddedCallbackList::iterator iter = myExecConnectedCbs.begin();
      iter != myExecConnectedCbs.end(); ++iter )
   {
      if ( -1 == ( *iter->first )( connName, connID, addr, hostName, fullyCompliant, iter->second ) )      
      {
         retVal = -1;
      }
   }
   return retVal;
}


inline void DtForwarderPort::addExecDisconnectedCb( DtExecRemovedChangeCb callback, void* usr )
{
   std::pair<DtExecRemovedChangeCb, void*> cbPair( callback, usr );
   DtExecRemovedCallbackList::iterator pos = std::find( myExecDisconnectedCbs.begin(),
      myExecDisconnectedCbs.end(), cbPair);   
   if ( pos == myExecDisconnectedCbs.end() )
   {
      myExecDisconnectedCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeExecDisconnectedCb( DtExecRemovedChangeCb callback, void* usr )
{
   DtExecRemovedCallbackList::iterator pos = std::find( myExecDisconnectedCbs.begin(),
      myExecDisconnectedCbs.end(), std::make_pair( callback, usr ));
   if ( pos != myExecDisconnectedCbs.end() )
   {
      myExecDisconnectedCbs.erase( pos );
   }
}

inline int DtForwarderPort::execDisconnected(const MAKRti::DtString& connName,
                     unsigned int connID, const DtInetAddrAndPort& addr, void* usr)
{
   return ((DtForwarderPort*)usr)->execDisconnected(connName, connID, addr);
}

inline int DtForwarderPort::execDisconnected( const MAKRti::DtString& connName,
                           unsigned int connID, const DtInetAddrAndPort& addr)
{
   int retVal = 0;
   for ( DtExecRemovedCallbackList::iterator iter = myExecDisconnectedCbs.begin();
      iter != myExecDisconnectedCbs.end(); ++iter )
   {
      if ( -1 == ( *iter->first )( connName, connID, addr, iter->second ) )      
      {
         retVal = -1;
      }
   }
   return retVal;
}

inline void DtForwarderPort::addWaitingForForwarderConnectionsCb(DtWaitingForFwsCb callback, void* usr)
{
   std::pair< DtWaitingForFwsCb, void* > cbPair( callback, usr );

   DtWaitingForFwsCallbackList::iterator pos = std::find( myWaitingForFwsCbs.begin(),
      myWaitingForFwsCbs.end(), cbPair );

   if( pos == myWaitingForFwsCbs.end() )
   {
      myWaitingForFwsCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeWaitingForForwarderConnectionsCb(DtWaitingForFwsCb callback, void* usr)
{
   DtWaitingForFwsCallbackList::iterator pos = std::find( myWaitingForFwsCbs.begin(),
      myWaitingForFwsCbs.end(), std::make_pair( callback, usr ));

   if( pos != myWaitingForFwsCbs.end() ) 
   {
      myWaitingForFwsCbs.erase( pos );
   }
}

inline void DtForwarderPort::waitingForForwarderConnections(const DtConnAndNameMap& allNeededFws,
                                                            const DtConnAndNameMap& stillNeededFws)
{
   for ( DtWaitingForFwsCallbackList::iterator iter = myWaitingForFwsCbs.begin();
      iter != myWaitingForFwsCbs.end(); ++iter )
   {
      ( *iter->first )( allNeededFws, stillNeededFws, iter->second );
   }
}

inline void DtForwarderPort::addNotWaitingForForwarderConnectionsCb(DtNotWaitingForFwsCb callback, void* usr)
{
   std::pair< DtNotWaitingForFwsCb, void* > cbPair( callback, usr );

   DtNotWaitingForFwsCallbackList::iterator pos = std::find( myNotWaitingForFwsCbs.begin(),
      myNotWaitingForFwsCbs.end(), cbPair );

   if( pos == myNotWaitingForFwsCbs.end() )
   {
      myNotWaitingForFwsCbs.push_back( cbPair );
   }
}

inline void DtForwarderPort::removeNotWaitingForForwarderConnectionsCb(DtNotWaitingForFwsCb callback, void* usr)
{
   DtNotWaitingForFwsCallbackList::iterator pos = std::find( myNotWaitingForFwsCbs.begin(),
      myNotWaitingForFwsCbs.end(), std::make_pair( callback, usr ));

   if( pos != myNotWaitingForFwsCbs.end() ) 
   {
      myNotWaitingForFwsCbs.erase( pos );
   }
}

inline void DtForwarderPort::notWaitingForForwarderConnections()
{
   for ( DtNotWaitingForFwsCallbackList::iterator iter = myNotWaitingForFwsCbs.begin();
      iter != myNotWaitingForFwsCbs.end(); ++iter )
   {
      ( *iter->first )( iter->second );
   }
}

inline int DtForwarderPort::socketQueueStatus( unsigned int connId, int connQueueStatus, void* usr )
{
    return ((DtForwarderPort*)usr)->socketQueueStatus( connId, connQueueStatus );
}

inline int DtForwarderPort::socketQueueStatus( unsigned int connId, int connQueueStatus )
{
    int retVal = 0;

    for( DtSocketQueueCallbackList::iterator iter = mySocketQueueCbs.begin();
        iter != mySocketQueueCbs.end(); ++iter )
    {
        if( -1 == ( *iter->first )( connId, connQueueStatus, iter->second ) )
        {
            retVal = -1;
        }
    }

    return retVal;
}


#endif