queueManager.h

Go to the documentation of this file.

/****************************************************************************
 * Copyright (c) 2012 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 DtPortalMessage;
   class DtPortalObject;
   class DtSubscribableMessageQueue;

   //
   //
   class DT_DLL_LIBPORTAL DtQueueManager
   {
   public:

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

   public:

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

      virtual ~DtQueueManager();

   public:

      virtual void send( DtPortalControlMessage& message );

      virtual void tick();

      virtual void requestBrokerShutdown();

      virtual void requestBrokerRestart( bool shouldMapMessagesOneToOne );

      virtual int connectedBrokerCount();

      virtual void pruneBrokerSocketMap();


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


   public:

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

   protected:

      virtual void addSocket( DtInetTcpSocket* sock );

      virtual void removeSocket( DtInetTcpSocket* sock );

      virtual void removeUnknownSocket( DtInetTcpSocket* sock );

      virtual void processUnknownSockets();

      virtual void testKnownSockets();

      void drainSockets();

   protected:

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

      virtual DtU8 getUnusedBrokerID();

   private:

      DtQueueManager( const DtQueueManager& other );

      DtQueueManager& operator = ( const DtQueueManager& other );

   protected:

      DtU32                         myControlSerialNumber;
      int                           myTickCountForSocketTest;
      int                           myTicksBeforeSocketTest;
      DtPortalMessageFactory        myMessageFactory;
      DtMemoryManager               myMemoryManager;
      DtSubscribableMessageQueue   *myControlMessageQueue;
      DtSubscribableMessageQueue   *myInteractionMessageQueue;
      DtSubscribableMessageQueue   *myObjectMessageQueue;
      DtInetSocketMgr               *myTcpSocketManager;
      DtClock                       myClock;
      DtTime                        myLastHeartbeatTime;

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

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

   };

}