rtiMessageForwarder.h

Go to the documentation of this file.

/*********************************************************************
** Copyright (c) 1992, 1997 MaK Technologies, Inc.
** All rights reserved.
*********************************************************************/
/*********************************************************************
** $RCSfile: rtiMessageForwarder.h,v $ $Revision: 1.19 $ $State: Exp $
*********************************************************************/
#ifndef DtRtiMsgForwarder_H_
#define DtRtiMsgForwarder_H_


#include "rtiMsConfig.h"
#include "RIDparams.h"
#include "multicastDiscoveryMgr.h"
#include "rtiVersion.h"
#include <vlutil/vlRunnable.h>
#include <vlutil/vlThreadSafe.h>
//#include "execThreadEvents.h"

#include <vlutil/vlSelectParamW.h>
#include <vlutil/vlSmartPointers.h>
#include <vlutil/vlInetAddr.h>

#include <utility>
#include <deque>

class DtRtiMsg;
class DtRtiMsgForwarder;
class DtFilteredTcpForwarder;
class DtMessageFilter;
class DtRtiExecTrackingFilter;
class DtFederationShadowTrackingFilter;
class DtRoutePrinterFilter;
class DtInternalRoutePrinterFilter;
class DtUpdateInterRoutingFilter;
class DtInternalRoutingFilter;
class DtFedexMsgFilter;
class DtMulitcastDiscoveryMgr;
class DtMsgFwdrFunctor;
class DtRtiMsgForwarder;
class DtConnectConfirmFilter;
class DtDeleteSocketFilter;
class DtTcpMsgSocket;
class DtTcpMsgSocketMgr;

namespace rti1516_2025
{
   class Authorizer;
}


typedef DtRtiMsgForwarder* (*DtRtiMsgFwderCreator)( unsigned long, const DtRIDParameters& params );

class DT_DLL_FORWARDER DtRtiMsgForwarder : public DtRunnable
{
protected:
   DtRtiMsgForwarder( unsigned long id, const DtRIDParameters& params );

public:
   
   struct DT_DLL_FORWARDER ConnectionInfo
   {
      ConnectionInfo( unsigned long connId, DtTcpMsgSocket* socket,
         const MAKRti::DtInetAddr& ipAddr, int port, 
         unsigned long federateHandle = 0,
         const MAKRti::DtString& federateName = "",
         const MAKRti::DtString& federateType = "",
         const MAKRti::DtString& fedexName = "",
         DtHlaVersionId hlaVersion = DtHla13);

      bool operator==( const ConnectionInfo& rhs );
      bool operator<( const ConnectionInfo& rhs );

      void setFederateInfo( unsigned long fedexHandle, unsigned long handle,
         const MAKRti::DtString& name, const MAKRti::DtString& type,
         const MAKRti::DtString& fedexName, DtHlaVersionId hlaVersion );
      void setConnectionId( unsigned long connId );

      bool myAuthorized;
      unsigned long myConnId;
      unsigned long myFederationHandle;
      unsigned long myHandle;
      unsigned long myProcessID;
      MAKRti::DtInetAddr myIp;
      int                myPort;
      MAKRti::DtString   myName;
      MAKRti::DtString   myType;
      MAKRti::DtString   myFederationName;
      MAKRti::DtString   myProcessName;
      MAKRti::DtString   myCommandArgs;
      DtHlaVersionId     myHlaVersion;
      const DtTcpMsgSocket*  mySocket;

      DtBitMask myFedIdMask;
   };

   typedef std::shared_ptr<ConnectionInfo> ConnectionInfoSP;
   typedef std::weak_ptr<ConnectionInfo>   ConnectionInfoWP;
   typedef std::map< const DtTcpMsgSocket*, ConnectionInfoSP >  SocketToConnInfoMap;
   

public:

   virtual ~DtRtiMsgForwarder();

   virtual void init();

   virtual bool hasExec() const;

   virtual void run();


   virtual void setEnableBundling( unsigned int maxBundleSize );

   virtual void setUseSmartForwarding( bool yesNo );

   virtual void setUseInternalSmartForwarding( bool yesNo );

   virtual void setRouteFedexMessages( bool yesNo );


   virtual void addPacketFilter( DtMessageFilter* filter );
   virtual void removePacketFilter( DtMessageFilter* filter );

   virtual inline unsigned long forwarderId() const;

   virtual void setForwarderId(unsigned long newId);

   virtual void updateTcpForwarderQueuedSockets();

   virtual rti1516_2025::Authorizer* authorizer();



   virtual inline const DtRtiMsgForwarder::SocketToConnInfoMap& getConnToConnInfoMap() const;

   virtual ConnectionInfoSP getConnectionInfoByConnId( unsigned long connId ) const;

   virtual ConnectionInfoSP getConnectionInfoByHandle( unsigned long federateHandle ) const;

   virtual ConnectionInfoSP getConnectionInfoBySocket( const DtTcpMsgSocket* sock ) const;

   virtual std::pair< bool, unsigned long > rtiExecConnected() const;

   virtual void sendMessage( const DtRtiMsg& msg, const DtTcpMsgSocket* src = 0 );

   virtual void requestCloseAllConnections();   

   virtual void requestKillSingleConnection( const DtTcpMsgSocket* sock );

   virtual void markConnectionAsAboutToBeDeleted(DtTcpMsgSocket const* socket, unsigned int connectionId);

   virtual bool isConnectionMarkedForDeletion(unsigned int connectionId) const;

   virtual void removeOldConnection(unsigned int connectionId);

   virtual void shutdownRequestedByExec();

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

   virtual void updateConnectionProcessParameters( unsigned long connId, bool authorized, unsigned long processId, MAKRti::DtString processName, MAKRti::DtString commandArgs );

   virtual bool shouldStop();

public:

   static DtRtiMsgForwarder* create( unsigned long id, const DtRIDParameters& params );

   static void setCreatorFunction(DtRtiMsgFwderCreator fn);

   static void tcpSocketAdditionCb(DtTcpMsgSocket* sock, void* usr);
   static void tcpSocketRemovalCb(DtTcpMsgSocket* sock, void* usr);

protected:

   virtual void performMcastDiscovery();

   virtual void applyRegisteredFilters();
   virtual void disableRegisteredFilters();

   virtual void waitForIo();


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

   virtual unsigned long consumeConnId();

   virtual void preRunLoopEvent() {};
   virtual void runLoopEvent() {};
   virtual void postRunLoopEvent() {};


   virtual bool populateConnectionInformation(
      unsigned int& numForwarderToForwarderConns,
      std::vector<MAKRti::DtString>& forwarderIdentifiers,
      std::vector<MAKRti::DtString>& forwarderConnectionHostNames,
      std::vector<MAKRti::DtInetAddr>& forwarderConnectionAddrs,
      unsigned int& numForwarderClientConns,
      std::vector<MAKRti::DtString>& clientConnectionHostNames,
      std::vector<MAKRti::DtInetAddr>& clientConnectionAddrs) const;

   virtual void closeAllConnectionsIfRequested();

   DtRtiMsgForwarder( const DtRtiMsgForwarder& orig );
   DtRtiMsgForwarder& operator=( const DtRtiMsgForwarder& orig );

protected:


   unsigned long myForwarderId;

   MAKRti::DtInetAddr myForwarderAddr;

   bool myCloseAllConnectionsRequested;

   DtFilteredTcpForwarder* myTcpForwarder;

   DtTcpMsgSocketMgr* mySocketMgr;

   int myPort;
   int myTcpNoDelay;

   unsigned long myNextConnectionIdOffset;

   unsigned int myPacketBundleSize;

   DtRtiExecTrackingFilter*    myRtiExecMonitor;
   DtFederationShadowTrackingFilter*    myFederationShadowMonitor;

   DtRoutePrinterFilter*       myRoutePrinter;
   DtInternalRoutePrinterFilter* myInternalRoutePrinter;
   DtUpdateInterRoutingFilter* myDmMsgRouter;
   DtInternalRoutingFilter*    myIntMsgRouter;

   DtFedexMsgFilter*           myFedexMsgRouter;

   DtConnectConfirmFilter* myConnectConfirmFilter;

   DtDeleteSocketFilter* mySocketMonitor;

   const DtRIDParameters& myParams;

   DtMulitcastDiscoveryMgr myMcastDiscoveryMgr;
   DtThreadSafe< std::deque<DtMsgFwdrFunctor*> > myPendingProxyCommands;

   std::vector< DtMessageFilter* > myRegisteredFilters;

   SocketToConnInfoMap myConnToConnInfoMap;

   std::set<DtTcpMsgSocket const*> myConnectionsAboutToBeDeleted;

   MAKRti::DtSelectParamWithWrite mySelectParams;

   bool myShouldStop;

   MAKRti::DtString myHostname;

   DtUniquePtr<rti1516_2025::Authorizer> myAuthorizer;

protected:

   static DtRtiMsgFwderCreator      theCreator;
};



inline const DtRtiMsgForwarder::SocketToConnInfoMap& DtRtiMsgForwarder::getConnToConnInfoMap() const
{
   return myConnToConnInfoMap;
}

inline unsigned long DtRtiMsgForwarder::forwarderId() const
{
   return myForwarderId;
}


#endif