inOutForwarderPort.h

Go to the documentation of this file.

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

#ifndef DtInOutForwarderPort_H_
#define DtInOutForwarderPort_H_

#include "forwarderPort.h"
#include "routeMap.h"
#include "rtiMsConfig.h"
#include "peerToPeerSockMgr.h"
#include <vlutil/vlTime.h>
#include <udpMsgSocket.h>
#include <map>

class DtRIDParameters;
class DtRouteMap;
class DtTcpSocketMgr__vl__;
class DtForwarderNodeInfoMsg;
class DtMacAddrMsg;
struct DtForwarderNodeInfo;

class DT_DLL_FORWARDER DtInOutForwarderPort : public DtForwarderPort
{

protected:

   DtInOutForwarderPort( const DtRIDParameters* params,
      std::shared_ptr<DtRouteMap> routes,
      const MAKRti::DtInetAddr& host,
      int forwarderPort,
      const MAKRti::DtInetEndpoint& forwarderAddrToConnectTo  );

public:

   virtual ~DtInOutForwarderPort();

   virtual bool isReady() const;

   virtual DtAddressMsgPair getNextMsg();

   virtual bool getInitializationMsgs( DtForwarderPort::DtSocketSet& socks,
      std::list<DtAddressMsgPair>& msgs, bool printBeDestMetadata);

   virtual bool getInitializationMsg( DtMsgSocket* sock,
      DtAddressMsgPair& msg, bool printBeDestMetadata);

   virtual void sendByTransport( const DtForwarderMsg& msg ) const;

   virtual void sendToAllReliable( const DtForwarderMsg& msg ) const;

   virtual void sendToListReliable( const DtForwarderMsg& msg,
      const DtAddressPortList& destinations ) const;

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

   virtual void updateRoutes();

   // Retrieve and process initialization messages if necessary.
   virtual void processInitializationMessages();

   // Process message acting on connection messages if necessary.
   virtual void processConnectionMessages(const DtAddressMsgPair& msgPair);

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

   // Attempts to fix the TCP connection.  Returns true if it was not fixed, 0 if it was successful.
   virtual bool fixConnection(DtTcpMsgSocket* sock);
   
   virtual void clearNeededConnections();

   virtual unsigned int abortPendingConnections();

   virtual unsigned int pendingConnectionsAborted();

   virtual DtAddressPortList connections() const;
   

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

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

   virtual bool flush();   
   
   virtual bool flushOnly();
   
   virtual MAKRti::DtString connectionListToString(const char* separator = " : ") const;

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

   virtual bool addPeerConnection(const MAKRti::DtString& host,
      const MAKRti::DtInetAddr& ipAddress, int port, bool distributeConnection);

   virtual void processMacAddrMsg(const DtForwarderPort::DtAddressMsgPair& msg);

   virtual void processPortMsg(const DtForwarderPort::DtAddressMsgPair& msgPair);

   virtual void processForwarderNodeInfoMsg(const DtForwarderPort::DtAddressMsgPair& msg);   

   virtual void processConnectToForwarderMsg(const DtForwarderPort::DtAddressMsgPair& msg);   

public:

   static DtForwarderPort* create( const DtRIDParameters* params,
      const std::shared_ptr<DtRouteMap> routes,
      const MAKRti::DtInetAddr& host, int forwarderPort,
      const MAKRti::DtInetEndpoint& additionalForwarderToConnectTo );

   static void setCreatorFunction( DtForwarderPortCreatorFcn creator );

protected:


   static void socketAdditionCb( DtTcpMsgSocket* sock, void* usr );
   virtual void socketAdditionCb( DtTcpMsgSocket* sock );


   static void socketEstablishedCb( DtTcpMsgSocket* sock, void* usr );
   virtual void socketEstablishedCb( DtTcpMsgSocket* sock );


   static void socketRemovalCb( DtTcpMsgSocket* sock, void* usr );
   virtual void socketRemovalCb( DtTcpMsgSocket* sock );

   virtual void initializeConnections();

   virtual void sendMacAddrMsg( DtTcpMsgSocket* sock );

   virtual void sendPortMsg( DtTcpMsgSocket* sock );

   virtual void sendForwarderNodeInfoMsgToConnections();

   virtual void sendForwarderNodeInfoMsg( DtTcpMsgSocket* sock, unsigned long nodeIdToUse=0);

   virtual void sendConnectToForwarderMsg(const DtInetAddrAndPort& fwDestination,
      const MAKRti::DtString& fwName, const MAKRti::DtString& fwHostName);

//    //! Returns whether sock was ok()
//    virtual bool reconnectToPeer( const DtInetAddrAndPort& addrToConnectTo, bool wasClient);
   
   virtual bool connectToPeer( const DtInetAddrAndPort& addrToConnectTo );
 
   virtual int processLostConnection( DtTcpMsgSocket* sock );

   virtual void connectionCompleted( const DtInetAddrAndPort& addrAndPort, DtTcpMsgSocket* sock );

   virtual void connectionCompleted(const MAKRti::DtString& nodeName, const MAKRti::DtString& hostName, unsigned long nodeId,
      const DtInetAddrAndPort& addrAndPort, DtTcpMsgSocket* sock);

   virtual void addSocketUsingRoutesFile(DtTcpMsgSocket* sock,
      const DtInetAddrAndPort& addrAndPort);

   virtual void addInitiatedSocketDynamically(DtTcpMsgSocket* sock, const DtInetAddrAndPort& addrAndPort);

   virtual void addReceivedSocketDynamically(DtTcpMsgSocket* sock, const DtInetAddrAndPort& addrAndPort);

   //Store / remove connection and update dependencies.
   virtual void addConnection( const DtInetAddrAndPort& newConn, DtTcpMsgSocket* sock );
   virtual void removeConnection( DtAddrToConnMap::iterator pos );

   // Return the socket with this listening destination
   virtual DtTcpMsgSocket* socketFromListeningAddr(const DtInetAddrAndPort& addrAndPort);

   virtual bool removeFromPendingList( DtTcpMsgSocket* addr );

   virtual void becomeRootForwarderNode();

   virtual void distributeConnections(const DtInetAddrAndPort& fwAddrAndPort,
      const MAKRti::DtString& fwName, const MAKRti::DtString& fwHostName, const std::set<DtForwarderNodeInfo>& fwConnPeers);

protected:

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

protected:


   bool                                             myIsReady;
   DtPeerToPeerTcpSockMgr*                          myConnectionMgr;
   DtForwarderPort::DtAddrToConnMap                 myConnections;   
   DtForwarderPort::DtAddrToConnMap::const_iterator myNextConnection;
   DtForwarderPort::DtAddrToConnMap::const_iterator myLastConnection;
   DtInetAddrAndPort                                myLocalDestination;
   DtInetAddrAndPort                                myPendingPeerAddition;

   DtAddrToConnMap                                  mySocketToBeRemoved;

   std::map<DtInetAddrAndPort, DtTcpMsgSocket*>     myAddrToSockMap;

   std::map<DtTcpMsgSocket*, unsigned long>         mySockToIdMap;

   DtForwarderPort::DtSocketSet                     myPendingConnections;
   DtForwarderPort::DtSocketSet                     myPrePendingConnections;
   DtForwarderPort::DtSocketSet                     myPostPendingConnections;
   DtForwarderPort::DtSocketSet                     myUnknownPortConnections;
   DtForwarderPort::DtSocketSet                     myDynamicallyInitiatedConnections;
   DtForwarderPort::DtSocketSet                     myDynamicallyReceivedConnections;

   std::map<DtInetAddrAndPort, MAKRti::DtString>            myNeededConnections;

   std::map<DtInetAddrAndPort, MAKRti::DtString>            myStillNeededConnections;

   std::map<DtInetAddrAndPort, DtInetAddrAndPort>   myUsedToListeningMap;
   std::map<DtInetAddrAndPort, DtInetAddrAndPort>   myListeningToUsedMap;


   DtForwarderPort::DtSocketList                    myInitiatedConnections;
   DtForwarderPort::DtSocketSet                     myExpiredConnections;
   DtForwarderPort::DtSocketSet                     myDiscardedRemoteSockets;

   DtUdpMsgSocket*                                  myLegacyUdpConnection;
   std::map<DtInetAddrAndPort, unsigned long long>  myMacAddrsOfConnectedForwardersMap;
   bool                                             myIsRootForwarderNode;

   MAKRti::DtClock                                  myClock;
   mutable std::map< DtTcpMsgSocket*, MAKRti::DtTime > mySktSendTimeMap;

   std::map< DtTcpMsgSocket *, unsigned int> myTestCountFailureMap;
   MAKRti::DtInetDevice myHostIf;

   int myCompressionLevel;

   MAKRti::DtString myHostname;

   unsigned long myMsgCount;

protected:

   static DtForwarderPortCreatorFcn theCreator;

};

inline bool DtInOutForwarderPort::updateFederateConnectionId( unsigned long orig, 
                                                              unsigned long final )
{
   return false;
}

#endif