extendedRtiMsgForwarder.h

Go to the documentation of this file.

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

#ifndef DtExtendedRtiMsgForwarder_H_
#define DtExtendedRtiMsgForwarder_H_

#include "forwarderPort.h"
#include "rtiMsConfig.h"
#include "rtiMessageForwarder.h"
#include "vlThreadedObject.h"
#include <vlutil/vlRunnable.h>
#include "vlThreadSafeQueue.h"
#include "vlThreadSafeBufferQueue.h"
#include <vlutil/vlTime.h>
#include <udpMsgSocket.h>
#include <set>

class DtRIDParameters;
class DtMessageBufferFilter;
class DtFederateTrackingFilter;
class DtNetworkMapLanRequestFilter;

class DT_DLL_FORWARDER DtExtendedRtiMsgForwarder : public DtRtiMsgForwarder
{
protected:

   DtExtendedRtiMsgForwarder( unsigned long id, const DtRIDParameters& params, bool postMessagesToParent = true );

public:

   typedef DtThreadSafeBufferQueue<char> DtCharBufferQueue;

   static DtExtendedRtiMsgForwarder* create( unsigned int id, const DtRIDParameters& params, bool postMessagesToParent = true );

   virtual ~DtExtendedRtiMsgForwarder();

   virtual void applyFilters();

   virtual void sendByTransport( const DtForwarderPort::DtForwarderMsg& msg );
   
   virtual void sendToAllReliable( const DtForwarderPort::DtForwarderMsg& msg );   

   virtual void sendBestEffort( const DtForwarderPort::DtForwarderMsg& msg );   

   virtual void sendMessageToWan( const DtRtiMsg& msg );

   virtual inline bool flush();

   virtual const DtCharBufferQueue::Buffer* getNextMsg();

   virtual void selectParams( MAKRti::DtSelectParamWithWrite& selectParams );

   virtual DtForwarderPort::DtAddressPortList newConnections() const;
   
   virtual DtForwarderPort::DtAddressPortList droppedConnections() const;
   
   virtual DtForwarderPort::DtAddressPortList connections() const;

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


   virtual void federateJoined(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);
   virtual void federateResigned(const MAKRti::DtString&,
      const MAKRti::DtString&, unsigned int, unsigned int,
      const DtInetAddrAndPort&  );
   virtual void updateSocketQueueStatus( unsigned int connId, int connQueueStatus);

   virtual void addFederateJoinedCb( 
      DtForwarderPort::DtFederateAddedCb callback, void* usr );
   virtual void addFederateResignedCb( 
      DtForwarderPort::DtFederateRemovedCb callback, void* usr ) ;

   virtual void addSocketAdditionCb(
      DtForwarderPort::DtConnectionAddRemoveCb cb, void* usr );
   virtual void addSocketRemovalCb(
      DtForwarderPort::DtConnectionAddRemoveCb cb, void* usr );

   virtual void addSocketQueueCb(
       DtForwarderPort::DtSocketQueueCb cb, void* usr );

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

   virtual void addExecConnectionLostCb( 
      DtForwarderPort::DtExecRemovedChangeCb callback, void* usr );
   virtual void removeExecConnectionLostCb( 
      DtForwarderPort::DtExecRemovedChangeCb callback, void* usr );

   virtual const DtFederateTrackingFilter* fedHandleTrackingFilter() const { return myFedHandleTracker; }

   
protected:

   virtual void init();

   virtual void tcpSocketAdditionCb(DtTcpMsgSocket* sock);
   virtual void tcpSocketRemovalCb(DtTcpMsgSocket* sock);

   virtual void refreshConnectionList() const;

public:
   
   static const unsigned int theForwarderId;

protected:

   typedef std::pair< DtInetAddrAndPort, int > DtAddressAndSockIdPair;

   DtUdpMsgSocket* myUdpRcvSocket;
   DtUdpMsgSocket* myUdpSendSocket;

   DtUniquePtr<DtCharBufferQueue::Buffer> myLastReceivedMsg;
   DtCharBufferQueue myInMsgQueue;
   DtCharBufferQueue myOutMsgQueue;
   DtMessageBufferFilter* myOutMsgFilter;
   DtFederateTrackingFilter*   myFedHandleTracker;

   DtClock myClock;

   MAKRti::DtTime myLastDiscoveryTickTime;

   DtForwarderPort::DtFederateAddedCallbackList myFederateJoinedCbs;
   DtForwarderPort::DtFederateRemovedCallbackList myFederateResignedCbs;

   DtForwarderPort::DtSocketQueueCallbackList mySocketQueueCbs;

   DtForwarderPort::DtConnectionAddRemoveCallbackList mySocketAddedCbs;
   DtForwarderPort::DtConnectionAddRemoveCallbackList mySocketRemovedCbs;
   DtForwarderPort::DtConnectionChangeCallbackList mySocketChangedCbs;

   typedef std::pair< bool, DtAddressAndSockIdPair > DtBoolAddressSocketIdPair;
   mutable DtThreadSafeQueue< DtBoolAddressSocketIdPair > mySocketListChanges;
   mutable std::set< DtAddressAndSockIdPair > myConnections;
   mutable DtForwarderPort::DtAddressPortList myNewConnections;
   mutable DtForwarderPort::DtAddressPortList myDroppedConnections;


   bool myForwardMessagesToParent;
};

inline bool DtExtendedRtiMsgForwarder::flush()
{
   bool status = true;
   if ( myUdpSendSocket->isOpen() )
   {
      status = myUdpSendSocket->flush();
   }

   return status;
}
#endif