queueManager.h

Go to the documentation of this file.

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


#pragma once

#include <portal/pMessageFactory.h>
#include <portal/pControlMessage.h>
#include <portal/memoryManager.h>

#include <vlutil/vlInetSocketMgr.h>
#include <vlutil/vlTime.h>

#include <list>
#include <map>

namespace MAKVrExchange
{

   class DtPortalObject;
   class DtPortalMessage;
   class DtSubscribableMessageQueue;

   //
   //
   class DT_DLL_LIBPORTAL DtQueueManager
   {
   public:

      typedef void (*MessageCbFcn)( const DtPortalMessage& message, void* usr );

   public:

      static DtQueueManager* create( DtU32 numberOfBucketsInQueue,
         DtU32 payloadSizePerBucket, int port, DtU16 theBrokerId,
         const DtString& objectQueueName, const DtString& interactionQueueName,
         const DtString& controlQueueName );

      DtQueueManager( DtU32 numberOfBucketsInQueue, DtU32 payloadSizePerBucket,
         int port, DtU16 brokerId, const DtString& objectQueueName,
         const DtString& interactionQueueName,
         const DtString& controlQueueName );

      virtual ~DtQueueManager();

      virtual void send( DtPortalControlMessage& message, bool global = false );

      virtual void tick();

      virtual void requestBrokerShutdown();

      virtual void requestBrokerRestart( bool shouldMapMessagesOneToOne );

      virtual void brokerShutdownNotification( DtU32 brokerId );

      virtual int connectedBrokerCount();

      virtual void pruneBrokerSocketMap();


      virtual DtSubscribableMessageQueue* getControlMessageQueue() { return myControlMessageQueue;}
      virtual DtSubscribableMessageQueue* getInteractionMessageQueue() { return myInteractionMessageQueue;}
      virtual DtSubscribableMessageQueue* getObjectMessageQueue() { return myObjectMessageQueue;}

      void processBrokerConnect(const DtBrokerStartedMessage& msg);

   protected:

      virtual void addSocket( DtInetTcpSocket* sock );

      virtual void removeSocket( DtInetTcpSocket* sock );

      virtual void removeUnknownSocket( DtInetTcpSocket* sock );

      virtual void processUnknownSockets();

      virtual void testKnownSockets();

      static void getLicFilesInDir( const std::string& dir, std::list<std::string>& files );

      static void searchForLicFile();

      void drainSockets();

      static bool addSocketCb( DtInetSocket* sock, void* usr );
      static bool removeSocketCb( DtInetSocket* sock, void* usr );

      virtual DtU8 getUnusedBrokerID();

   private:

      DtQueueManager( const DtQueueManager& other );

      DtQueueManager& operator = ( const DtQueueManager& other );

   protected:

      const DtU16 myBrokerId;

      DtU32 myControlSerialNumber;
      int myTickCountForSocketTest;
      int myTicksBeforeSocketTest;

      DtPortalMessageFactory myMessageFactory;
      DtMemoryManager myMemoryManager;

      DtSubscribableMessageQueue* myControlMessageQueue;
      DtSubscribableMessageQueue* myInteractionMessageQueue;
      DtSubscribableMessageQueue* myObjectMessageQueue;

      DtInetSocketMgr* myTcpSocketManager;

      DtClock myClock;
      DtTime myLastHeartbeatTime;

      typedef std::set<DtU32> BrokerShutdownNotificationSet;
      BrokerShutdownNotificationSet myBrokerShutdownNotifications;

      typedef std::map<int,DtInetTcpSocket*> BrokerSocketMap;
      BrokerSocketMap myBrokerSocketMap;

      typedef std::map<DtInetTcpSocket*,int> UnknownSocketMap;
      UnknownSocketMap myUnknownBrokerSocketList;

      typedef std::map<int, int> BrokerIdTypeMap;
      BrokerIdTypeMap myBrokerIdTypeMap;

   };

}