asyConMgr.h

Go to the documentation of this file.

/*******************************************************************************
** Copyright (c) 1998 MaK Technologies, Inc.
** All rights reserved.
*******************************************************************************/
/*******************************************************************************
** $RCSfile: asyConMgr.h,v $ $Revision: 1.39 $ $State: Exp $
*******************************************************************************/


#ifndef VXWORKS

#ifndef DtAsyncConnectionMgr_H_
#define DtAsyncConnectionMgr_H_


#include "rtiMsConfig.h"
#include "connectMgr.h"
#include <vlutil/vlThreadSafe.h>
#include <vlutil/vlMutex.h>
#include "fedAmbCallbackQueue.h"

#ifdef _WIN32
#include "windows.h"
#else
#include "pthread.h"
#endif

#ifdef DtIFSPEC13DLC
#define RTI rti13
#endif

namespace MAKRti
{
   class DtSelectParamWithWrite;
}
class DtFedAmbCallbackMsg;

class DT_DLL_LRC DtQueuePacket
{
public:
   DtQueuePacket( DtRtiMsg* msg,
                  MAKRti::DtTransportType transport,
                  const MAKRti::DtInetAddr& addr = MAKRti::DtInetAddr::inaddrAny() );

   ~DtQueuePacket();

   DtRtiMsg* msg;
   MAKRti::DtInetAddr addr;
   MAKRti::DtTransportType transport;
};


class DT_DLL_LRC DtAsyncConnectionMgr : public DtConnectionMgr
{
protected:

   DtAsyncConnectionMgr( DtRIDParameters* params, bool isFederate = false, 
      DtBoost::shared_ptr<DtRtiAssistantLrcClient> assistantClient = 
      DtBoost::shared_ptr<DtRtiAssistantLrcClient>() );

public:

   virtual ~DtAsyncConnectionMgr();

   virtual int init( DtRIDParameters* params );

   virtual void setFederateMgr( DtFederateMgr* fedMgrPtr );

   virtual void shutdown();

   virtual bool isAsynchronous();

   virtual void disconnectFederation();
   virtual void connectFederation(
#ifdef DtIFSPEC1516
      rti1516::FederateAmbassador* fedAmb,
#elif defined(DtIFSPEC13)
      RTI::FederateAmbassador* fedAmb,
#endif
      DtLogicalTimeFactory* logicalTimeFactoryPtr,
      MAKRti::DtString fedExName, 
      DtFederateHandle fedHandle,
      int fedExHandle );

   virtual void queueMsg( DtRtiMsg* msg, 
                          MAKRti::DtTransportType transport,
                          const MAKRti::DtInetAddr& addr = MAKRti::DtInetAddr::inaddrAny() ); 

   virtual int send( const DtRtiMsg& msg, 
                     MAKRti::DtTransportType transport,
                     const MAKRti::DtInetAddr& addr = MAKRti::DtInetAddr::inaddrAny() );

   virtual int sendAndDelete( DtRtiMsg* msg, 
                              MAKRti::DtTransportType transport,
                              const MAKRti::DtInetAddr& addr = MAKRti::DtInetAddr::inaddrAny() );

   virtual bool tick();

   virtual bool processConnections();

   virtual DtRtiMsg* getMessage();

   virtual DtNetReadStatus handleMsg( DtRtiMsg& msg );

   virtual bool flush();

   virtual void workerThread();

   virtual bool wait( MAKRti::DtTime periodInSeconds, bool unlockWhileWaiting = false );

#ifdef DtIFSPEC1516
   virtual int readFileDescriptor(rti1516::TransportationType transport);
#elif defined(DtIFSPEC1516E)
   virtual int readFileDescriptor(rti1516e::TransportationType transport);
#else
   virtual int readFileDescriptor(RTI::TransportType transport);
#endif

#ifdef _WIN32

   virtual HANDLE waitEvent();
#ifdef DtIFSPEC1516
   virtual WSAEVENT readFileEvent(rti1516::TransportationType transport);
#elif defined(DtIFSPEC1516E)
   virtual WSAEVENT readFileEvent(rti1516e::TransportationType transport);
#else
   virtual WSAEVENT readFileEvent(RTI::TransportType transport);
#endif
#else

   virtual int waitEvent();
#endif

   virtual void signalReceiveEvent();

   virtual void clearReceiveEvent();

   virtual void signalSendEvent();

   virtual void clearSendEvent();

   virtual void yield();

protected:

   virtual void doDisconnectFederation();

   virtual bool doWait(MAKRti::DtTime period);

   virtual void processCallbackMsg(const DtFedAmbCallbackMsg& callbackMsg);

   virtual DtQueuePacket* getPacket();

   virtual bool sendPackets();

   virtual DtNetReadStatus receivePacket(DtRtiConnection* conn,
      MAKRti::DtTransportType transportType, unsigned int& receivedPacketCount);

   virtual bool receivePackets();

   virtual DtNetReadStatus processMsgQueueCallbacks(DtRtiMsg& msg,
      unsigned int& receivedPacketCount);

   virtual void processPendingMsgQueueCallbacks();

   virtual void processPendingMsgMakeCallbacks();

   virtual void setMessageTracking(DtRtiMsg& msg);

#ifdef _WIN32
   virtual void waitOnIO();
#else
   virtual void waitOnIO(DtSelectParamWithWrite& params);
#endif

   virtual void destroyPackets();

   virtual void destroyRelConn();   

   virtual void destroyAncillaryRelConn(DtRtiConnection* conn);

   virtual void destroyAncillaryRelConns();

   virtual void startAsynchronousProcessing();

   virtual void stopAsynchronousProcessing();

public:
 
   static DtConnectionMgr* create( DtRIDParameters* p,  bool isFederate = false, 
      DtBoost::shared_ptr<DtRtiAssistantLrcClient> assistantClient = 
      DtBoost::shared_ptr<DtRtiAssistantLrcClient>() );

#ifdef _WIN32
   static unsigned long run( void* usr );
#else
   static void* run( void* usr );
#endif

protected:

   static void processCallbackMsg(const DtFedAmbCallbackMsg& callbackMsg, void* usr);

protected:
   bool myTimeToQuit;
   bool myThreadActivated;
   bool myAsynchronous;
   bool myAsynchronousCallbacks;
   bool myAsynchronousProcessMsg;
   bool myReserved;
   bool myMoreToFlush;
   bool myDoDisconnect;

   DtMutex* mySendLock;
   DtMutex* myReceiveLock;

#ifdef _WIN32
   HANDLE myThreadHandle;
   HANDLE myEvent;
   HANDLE*      myWaitEvents;
   unsigned int myNumWaitEvents;
#else
   pthread_t myThreadHandle;
   int mySendEvent[2];
   int myRecvEvent[2];
#endif

   float myIoPeriod;
   float myTickWaitPeriod;
   unsigned int myMaxQueue;
   unsigned int myMaxCount;
   unsigned int myLockQueue;
 
   long myReceivedBufferCurrent;
   long myReceivedBufferCount;
   DtQueuePacket** myReceivedBuffer;

   unsigned int myCurrentSendPacket;
   unsigned int myPacketsToSendCount;
   std::vector<DtQueuePacket*> mySendPacketBuffer;
   std::vector<DtQueuePacket*> myRecvPacketBuffer;

   unsigned int myReadPacketCount;

   MAKRti::DtClock      myBestEffortRetryTimer;
   unsigned int myBestEffortRetryAttempts;
   MAKRti::DtTime       myBestEffortTimeToRetry;      

   MAKRti::DtList* mySendQueue;
   MAKRti::DtList* myReceiveQueue;

   DtCallbackQueue* myCallbackQueue;

   DtThreadSafe<bool> myAsynchronousConnectionIsOk;

   unsigned int myNextReliableConnToRead;


};

#endif   

#endif