rtiAssistantPeer.h

Go to the documentation of this file.

/****************************************************************************
 * Copyright (c) 2016 VT MAK
 * All rights reserved.
 ****************************************************************************/


#pragma once

#include <rtiUtil.h>
#include <rtiVersion.h>

#include "assistantDefaultMsgFact.h"
#include "assistantMsgCbMgr.h"
#include "assistantLicServerInfo.h"
#include "assistantConnectionInfo.h"
#include "assistantConnMsg.h"
#include "connectionAndConfiguration.h"
#include "rtiAssistantClient.h"
#include "assistantInfo.h"
#include "assistantInfoMsg.h"
#include "assistantNetStatsStringReqMsg.h"
#include "assistantNetStatsEnums.h"
#include "assistantToAssistantConnectionMsg.h"
#include <rtiutil/tcpMsgSocketMgr.h>

#include <vlutil/vlSmartPointers.h>
#include <vlutil/vlInetEndpoint.h>
#include <vlutil/vlPrint.h>
#include <vlutil/vlTime.h>
#include <vlutil/vlUtil.h>

class DtAssistantQueryMsg;
class DtAssistantInfoMsg;
class DtAssistantRemovalMsg;
class DtAssistantLightweightInfoMsg;
class DtAssistantConfig;
class DtAssistantFederationInfo;
class DtRtiConnectionConfigurationInfo;
class DtUdpMsgSocket;
class DtAssistantFederateInfo;
class DtAssistantStartRtiexecMsg;
class DtAssistantPeerSocketAdditionCbInfo;
class DtAssistantPeerSocketRemovalCbInfo;
typedef void (*DtInterfaceFoundCb)( void* );
#define DtCbInfoStruct( cbType )                 \
struct cbType##Info                              \
{                                                \
   cbType##Info( cbType cb, void* usr )  :       \
      myCb(cb),                                  \
      myUsr(usr)                                 \
      { };                                       \
   cbType myCb;                                  \
   void* myUsr;                                  \
};


typedef void(*DtAssistantConnectionLostCallback)(const DtAssistantConnectionStruct& sock, void* usr);
class DtAssistanConnectionCbInfo
{
public:

   DtAssistanConnectionCbInfo(DtAssistantConnectionLostCallback cb, void* usr)
   {
      myCb = cb;
      myUsr = usr;
   }

   DtAssistantConnectionLostCallback myCb;
   void* myUsr;
};

typedef void(*DtAssistantSocketAdditionCb)(int compId, void* usr);
typedef void(*DtAssistantSocketRemovalCb)(int compId, const MAKRti::DtString& reason, void* usr);
typedef std::list<DtAssistantPeerSocketAdditionCbInfo*> DtAssistantPeerSocketAdditionCallbackList;
typedef std::list<DtAssistantPeerSocketRemovalCbInfo*> DtAssistantPeerSocketRemovalCallbackList;
class DT_DLL_ASSISTANT DtRtiAssistantPeer
{
public:

   DtRtiAssistantPeer( int id, int port,
      const MAKRti::DtInetAddr& ipv4Addr,
      const MAKRti::DtInetAddr& ipv6Addr,
      int ttl,
      bool disableMulticast,
      const DtNetU128 & uuid);

   virtual ~DtRtiAssistantPeer();

   virtual bool tick( double timeout = 0 );

   virtual void setMsgFactory( DtAssistantMessageFactory* fact );

   virtual DtAssistantMessageFactory* msgFactory() const;


   virtual void addMsgCallback( DtAssistantMsgKind msgKind,
      DtAssistantMsgCbMgr::DtCallbackFcn fcn, void* usr );
   virtual void removeMsgCallback( DtAssistantMsgKind msgKind,
      DtAssistantMsgCbMgr::DtCallbackFcn fcn, void* usr );

   virtual void addInterfaceFoundCallback( DtInterfaceFoundCb cb, void* usr );
   virtual void removeInterfaceFoundCallback( DtInterfaceFoundCb cb, void* usr );


   virtual void sendQueryMsg();

   virtual void sendForeignQueryMsg(const DtAssistantInfo& assistant);

   virtual void sendForeignQueryMsg(const DtAssistantConnectionStruct& assistant);

   virtual void sendForeignBreakMsg(const DtAssistantInfo& assistant);

   virtual void sendAssistantConnMsg(const DtAssistantConnectionStruct& assistant, DtAssistantHandshakeType handshakeType, const DtAssistantConnectionId& id, const DtAssistantConnectionId& ackId = DtAssistantUuidNIL);

   virtual void sendInfoMsg( const DtRtiConnectionAndConfiguration& connsAndConfigs,
      unsigned memoryPeriod, bool acceptRemoteCommands, bool removal );

   virtual void sendLightweightInfoMsg( const DtRtiConnectionAndConfiguration& connsAndConfigs,
      unsigned memoryPeriod );

   virtual void sendLicServerInfoMsg( const DtLicServerSet& licServers,
      bool removal );

   virtual void sendForwarderConnectMsg( int compId, const DtString& targetHost,
      const MAKRti::DtInetAddr& targetAddress, int targetPort,
      const DtAssistantInfo& assistant );

   virtual void sendStartRtiForwarderMsg( const DtRtiConnectionInfoDataPair& conn,
      const DtRtiConnectionConfigurationInfo& config,
      const DtAssistantInfo& assistant = DtAssistantInfo() );

   virtual void sendStartRtiexecMsg( const DtRtiConnectionInfoDataPair& conn,
      const DtRtiConnectionConfigurationInfo& config,
      const DtAssistantInfo& assistant = DtAssistantInfo() );

   virtual void sendObserveFederationsMsg( const DtRtiConnectionInfo& conn,
      bool observe, const DtAssistantInfo& assistant = DtAssistantInfo(),
      int compId = -1 );

   virtual void sendObserveRtiExecLogMsg( const DtAssistantInfo& assistant,
      int compId, bool observe );

   virtual void sendObserveRtiexecRspMsg( DtNotifyLevelType notifyLevel, int compId,
      const DtAssistantInfo& destination );

   virtual void sendObserveRtiForwarderMsg( const DtRtiConnectionInfo& conn,
      bool observe, const DtAssistantInfo& assistant, int compId );

   virtual void sendFederationStateMsg( const DtRtiConnectionInfoDataPair& conn,
      const DtAssistantFederationInfo& federation, bool isRemoval );

   virtual void sendNoFederationMsg( const DtRtiConnectionInfoDataPair& conn );

   virtual void sendFederateResignMsg( const DtAssistantInfo& assistant,
      int compId, int federationHdl, int federateHdl );

   virtual void sendFederationDestroyMsg( const DtAssistantInfo& assistant,
      int compId, int federationHdl );

   virtual void sendRidParamsRequestMsg( const DtAssistantInfo& assistant,
      int compId );

   virtual void sendLogRequestMsg( const DtAssistantInfo& assistant,
      int compId, unsigned maxLogSize );

   virtual bool sendLatencyTestRequestMsg( const DtAssistantInfo& assistant,
      int compId, int federationHdl, int federateHdl, bool runTest, bool continuous,
      bool shouldHalt, bool observing, unsigned int interactionPeriod,
      unsigned int numInteractions, unsigned int payloadBytes );

   virtual void sendNetworkTestEnableMsg( const DtAssistantInfo& assistant,
      int compId, int federationHdl, bool enable );

   virtual void sendRtiexecExitMsg( const DtAssistantInfo& assistant, int compId );

   virtual void sendNotifyObserveMsg( bool isObserving );

   virtual void sendNotifyMsg( const MAKRti::DtString& text,
      DtNotifyLevelType notifyLevel, const MAKRti::DtString& sourceName );

   virtual void sendRemovalMsg();

   virtual void sendFederateInfoReqMsg(const DtAssistantInfo& astnt,
      const DtRtiConnectionInfo& conn,
      const MAKRti::DtString& federation = MAKRti::DtString::nullString(),
      unsigned federateHandle = 0 );

   virtual void sendFederateInfoRespMsg(const DtRtiConnectionInfo& conn,
      std::vector<const DtAssistantFederateInfo*> federates);


   virtual void sendNetStatsObservationRequestMsg(const DtAssistantInfo& astnt, DtAssistantObservationType observationType,
      int compId, bool observe = false, unsigned int interval = 0);

   virtual void sendNetStatsObservationResponseMsg(const DtAssistantInfo& astnt, DtAssistantObservationType observationType,
      int compId, bool observe);

   virtual void sendNetStatsStringUpdateReqMsg(const DtAssistantInfo& assistant, int compId,
      DtAssistantHandleType handletype, const std::set<DtU32>& handles);

   virtual void sendNetStatsStringUpdate(const DtAssistantInfo& assistant, int compId,
      DtAssistantHandleType handletype, const std::vector<DtU32>& objectInstanceHandles,
      const std::vector<MAKRti::DtString>& objectInstanceNames);

   virtual void forwardNetStatsUpdate(const DtAssistantMsg &msg, const std::set<DtAssistantInfo>& observers);

   virtual void sendFedDisconnectNetStatsObsRespMsg(DtAssistantObservationType observationType,
      int compId, const std::set<DtAssistantInfo>& observers);

   virtual void sendNetStatsInitStateRequestMsg(const DtAssistantInfo& astnt,
      DtAssistantObservationType observationType, int compId);

   virtual void sendFomTreeData(const DtAssistantInfo& astnt,
      int compId, const std::vector<MAKRti::DtU32>& objectClassHandles,
      const std::vector<MAKRti::DtU32>& parentObjectClassHandles,
      const std::vector<MAKRti::DtString>& objectClassNames,
      const std::vector<MAKRti::DtU32>& interactionClassHandles,
      const std::vector<MAKRti::DtU32>& parentInteractionClassHandles,
      const std::vector<MAKRti::DtString>& interactionClassNames);

   virtual void sendNetStatsAttrStringUpdate(const DtAssistantInfo& astnt,
      int compId, const std::vector<MAKRti::DtU32>& objectClassHandles,
      const std::vector<MAKRti::DtU32>& attributeHandles,
      const std::vector<MAKRti::DtString>& attributeNames,
      const std::vector<MAKRti::DtU32>& parentClassHandles,
      const std::vector<MAKRti::DtU32>& totalObjHandles);
   virtual void sendNetStatsParamStringUpdate(const DtAssistantInfo& astnt,
      int compId, const std::vector<MAKRti::DtU32>& interClassHandles,
      const std::vector<MAKRti::DtU32>& parameterHandles,
      const std::vector<MAKRti::DtString>& parameterNames,
      const std::vector<MAKRti::DtU32>& parentInterHandles,
      const std::vector<MAKRti::DtU32>& totalInterHandles);
   virtual void sendNetStatsAttrParamStringUpdate(const DtAssistantInfo& astnt,
      int compId, const std::vector<MAKRti::DtU32>& objectClassHandles,
      const std::vector<MAKRti::DtU32>& attributeHandles,
      const std::vector<MAKRti::DtString>& attributeNames,
      const std::vector<MAKRti::DtU32>& parentClassHandles,
      const std::vector<MAKRti::DtU32>& totalObjHandles,
      DtAssistantHandleType objInterType, DtAssistantHandleType attrParamType,
      DtAssistantHandleType parentClassType );
   virtual void sendNetStatsFomAttrParamUpdate(const DtAssistantInfo& astnt,
      int compId, DtAssistantHandleType handleType, bool isInit,
      const std::vector<MAKRti::DtU32>& objectClassHandles,
      const std::vector<MAKRti::DtU32>& attributeHandles,
      const std::vector<bool>& attributePub,
      const std::vector<bool>& attributeSub,
      const std::vector<DtTransportType>& attributeTransport,
      const std::vector<DtOrderType>& attributeOrder,
      const std::vector<bool>& objectPub, const std::vector<bool>& objectSub,
      const std::vector<MAKRti::DtU32>& objPubSubHandles );

   virtual void forwardMsg( const DtAssistantMsg& msg, int astntCompId,
      const DtAssistantInfo& dest = DtAssistantInfo() );

   virtual int port() const;

   virtual MAKRti::DtInetAddr ipv4Address() const;

   virtual MAKRti::DtInetAddr ipv6Address() const;

   virtual int id() const;

   virtual const MAKRti::DtInetAddr& currentNetworkInterfaceAddr() const;

   virtual const DtInetEndpoint& currentReceivedEndpoint() const;

   virtual const MAKRti::DtInetAddr& destinationAddress(const DtInetAddr& msgDestination) const;

   virtual void getMsgStats( unsigned& totalRecv, double& recvPerSecOverall,
      double& recvPerSecCurr, unsigned& totalProc, double& procPerSecOverall,
      double& procPerSecCurr, unsigned& totalSent, double& sentPerSecOverall,
      double& sentPerSecCurr );

   virtual DtU32 maxMsgSize() const;

   void sendMsgToAssistant( const DtAssistantInfo& ifAddr,
      DtAssistantMsg& msg );

   virtual void addForeignAssistantConnection(const DtAssistantConnectionStruct& ipAddr, const MAKRti::DtInetAddr& ifAddr);

   virtual void removeForeignAssistantConnection(const DtAssistantConnectionStruct& ipAddr);

   virtual void setUuid(const DtAssistantUuid& uuid);
   virtual const DtAssistantUuid& uuid() const;

   virtual const DtHostNameAddress& hostNameAddress() const;

#ifdef DtDIAGNOSTIC

   void enableMsgDiagnostics( bool enable );

#endif

   virtual void setIgnoreAllPeerCommunication(bool valToSet) { myIgnoreAllPeerCommunication = valToSet; };
   virtual bool isIgnoreAllPeerCommunication() const { return myIgnoreAllPeerCommunication; };

   virtual void addIgnoredAddress(const MAKRti::DtInetAddr& ipAddr);

   virtual void removeIgnoredAddress(const MAKRti::DtInetAddr& ipAddr);

   void addConnectionLostCallback(DtAssistantConnectionLostCallback cb, void* usr);
   void removeConnectionLostCallback(DtAssistantConnectionLostCallback cb, void* usr);

protected:

   virtual void sendStartRtiexecOrForwarderMsg( DtAssistantMsgKind msgKind,
      const DtRtiConnectionInfoDataPair& conn, const DtRtiConnectionConfigurationInfo& config,
      const DtAssistantInfo& assistant = DtAssistantInfo() );

   bool needsProcessing( const DtAssistantMsg& msg );

   bool addSocketsForInterfaces();

   bool addTcpServerSocketsForInterfaces(const DtTime& now = 0);

   bool readAndProcess( DtTime maxTime );

   bool readAndProcessConnectingTcpSockets(DtTime maxTime);

   bool readAndProcessTcpSockets(DtTime maxTime);
   void retransmitToForeignConnections(const DtAssistantMsg& msg);
   void checkForDeadTcpConnections();
   void calculateMsgStatistics();

   void sendMsgOnInterface( const MAKRti::DtInetAddr& ifAddr,
      DtAssistantMsg& msg, const DtInetAddr& destAddr = DtInetAddr::inaddrAny(), int port = 0);

   void sendMsg( std::map< MAKRti::DtInetAddr, DtUdpMsgSocket*>::iterator sockIter,
      DtAssistantMsg& msg, const DtInetAddr& destAddr = DtInetAddr::inaddrAny(), int port = 0, bool sendToForeignAssistants = true );

   void sendToForeignConnection(std::map< DtInetAddr, DtUdpMsgSocket*>::iterator sockIter,
      DtAssistantMsg& msg, const DtInetAddr& destAddr, int port);
   void sendToUDPForeignConnection(std::map< DtInetAddr, DtUdpMsgSocket*>::iterator sockIter, DtAssistantMsg* ptToPtMsg, const DtInetAddr& destAddr, int port);
   void sendToTCPForeignConnection(std::map< DtInetAddr, DtUdpMsgSocket*>::iterator sockIter, DtTcpMsgSocket* tcpSock, DtAssistantMsg* ptToPtMsg);
   void sendMsgToAll( DtAssistantMsg& msg, const DtInetAddr& destAddr = DtInetAddr::inaddrAny(), int port = 0, bool sentToForeignAssistants = true );

   virtual void flushTcpSockets(DtTime maxTime);

   DtUniquePtr<DtAssistantMsg> netRead(DtMsgSocket* sock, int *retCode, DtInetEndpoint& ep) const;

   virtual void sendTcpMsg(const DtAssistantConnectionStruct& connection, DtAssistantHandshakeType handshakeType, DtAssistantToAssistantConnectionMsg* msgToSend);
   virtual DtTcpMsgSocket* initiateTcpConnection(const DtAssistantConnectionStruct& connection);
   virtual void queueTcpSynMessage(DtTcpMsgSocket* sock, DtAssistantToAssistantConnectionMsg* msgToSend);
   virtual void addConnectingSocket(DtTcpMsgSocket* sock);
   virtual void deletePendingSyns(DtTcpMsgSocket* sock);
   virtual void processConnections();
   virtual void reset(const DtAssistantConnectionStruct& connection);
   static void tcpSocketAdditionCb(DtTcpMsgSocket* sock, void* usr);
   virtual void tcpSocketAdditionCb(DtTcpMsgSocket* sock);
   static void tcpSocketRemovalCb(DtTcpMsgSocket* sock, void* usr);
   virtual void tcpSocketRemovalCb(DtTcpMsgSocket* sock);


   virtual void invokeRemovalCallbacks(const DtAssistantConnectionStruct& second);
   virtual void removeSocket(DtTcpMsgSocket* sock, const DtAssistantConnectionStruct& conn);
public:

   DtCbInfoStruct(DtInterfaceFoundCb)

protected:

   int myPort;

   MAKRti::DtInetAddr myIpv4Addr;

   MAKRti::DtInetAddr myIpv6Addr;

   int myTtl;

   int myId;

   std::map< MAKRti::DtInetAddr, DtUdpMsgSocket*> mySockets;

   std::map< MAKRti::DtInetAddr, DtRemoteAssistantSet> myInterfaceToUdpForeignConnections;

   // TODO maybe this compare is wrong (should be by the tok?)
   std::map< DtAssistantConnectionStruct, MAKRti::DtInetAddr, cmpByPublicAddress> myUdpForeignConnectionToInterface;

   std::map< DtAssistantConnectionStruct, std::pair< DtTcpMsgSocket*, MAKRti::DtInetAddr >, cmpByConnectionId > myTcpForeignConnections;
   std::set<DtTcpMsgSocket*> myServerSocketsToDelete;
   std::vector<std::pair<DtAssistantConnectionStruct, DtTime> > myClientSocketsDeleted;
   std::map<DtTcpMsgSocket*, DtAssistantConnectionStruct> myAssistantConnectionsRemoved;
   std::map< MAKRti::DtInetAddr, DtUdpMsgSocket*>::iterator mySocketIter;

   DtTcpMsgSocket* myCurTcpSocket;

   MAKRti::DtInetAddr myCurrNetworkInterfaceAddr;

   MAKRti::DtInetEndpoint myCurrReceivedEndpoint;

   DtAssistantMsgCbMgr myMsgCbMgr;

   DtAssistantMessageFactory* myMessageFactory;

   DtAssistantDfltMsgFactory myDefaultMessageFactory;

   DtClock myClock;

   DtTime myNextCheckInterfacesTime;

   DtTime myNextClearCheckedInterfacesTime;
   
   DtTime myNextCheckDeadTcpConnections;
   
   DtTime myCheckInterfacesInterval;

   DtTime myClearCheckedInterfacesInterval;
   
   DtTime myCheckTcpConnectionsInterval;

   DtTime myNextProcessConnectionsTime;

   DtTime myNextProcessConnectionsInterval;

   DtTime myBeginProcessingTime;

   DtTime myNextStatsTime;

   DtTime myLastStatsTime;

   struct DtPeerMsgStats
   {
      DtPeerMsgStats() :
         numMsgsCurrPeriod(0), numMsgsTotal(0),
         msgsPerSecLastPeriod(0), msgsPerSecOverall(0) {}

      unsigned numMsgsCurrPeriod;
      unsigned numMsgsTotal;
      double msgsPerSecLastPeriod;
      double msgsPerSecOverall;
   };

   DtPeerMsgStats myReceivedMsgStats;

   DtPeerMsgStats myProcessedMsgStats;

   DtPeerMsgStats mySentMsgStats;

   std::list<DtInterfaceFoundCbInfo> myInterfaceFoundCallbacks;

   static const DtTime theStatisticsInterval;

   bool myUsingIpv4;

   bool myUsingIpv6;

   unsigned long mySequence;

   bool myDisableMulticast;

   bool myIgnoreAllPeerCommunication;

   std::set<DtInetAddr> myIgnoredAddresses;

   DtAssistantUuid myUuid;

#ifdef DtDIAGNOSTIC
   bool myDiagnosticsEnabled;
#endif

   DtString myHostname;
   DtHostNameAddress myHostNameAddress;
   struct cmpByMacAddr {
      bool operator()(const DtInetDevice& a, const DtInetDevice& b) const {
         return a.macAddrString() < b.macAddrString();
      }
   };
   // This comparison is currently expensive, but its sort of a rare cost as we don't expect sockets map to grow too big (most computers have 2 network cards)
   struct cmpByMacAddrThenFam {
      bool operator()(const std::pair<DtString, DtInetUtils::DtInetAddrFamily>& a, const std::pair<DtString, DtInetUtils::DtInetAddrFamily>& b) const {
         if (a.first != b.first)
         {
            return a.first < b.first;
         }
         else
         {
            return a.second < b.second;
         }
      }
   };
   typedef std::map<std::pair<DtString, DtInetUtils::DtInetAddrFamily>, DtTcpMsgSocketMgr*, cmpByMacAddrThenFam> DtTcpServerMap;
   DtTcpServerMap myPeerTcpServerSockets;
   bool myProcessingConnections;
   std::map<DtTcpMsgSocket*, MAKRti::DtTime> myPendingConnections;
   
   unsigned int myPendingConnectionsTimeout;

   std::map<DtTcpMsgSocket*, DtAssistantToAssistantConnectionMsg*> myPendingTcpSynMessages;

   //std::map<DtTcpMsgSocket*, DtAssistantConnectionId> myConnectingTcpClientSockets;
   std::map<DtAssistantConnectionId, DtTcpMsgSocket*> myConnectingTcpClientSockets;
   std::set<DtTcpMsgSocket*> myConnectingTcpServerSockets;
   DtAssistantPeerSocketAdditionCallbackList myTcpClientAdditionCallbacks;
   DtAssistantPeerSocketRemovalCallbackList myTcpClientRemovalCallbacks;


   std::set<DtAssistanConnectionCbInfo*> myConnectionLostCallbacks;

   std::set<std::pair<DtString, DtInetUtils::DtInetAddrFamily>, cmpByMacAddrThenFam> myDevicesChecked;
};