smSubscribableMessageQueue.h

Go to the documentation of this file.

/*********************************************************************
  ** Copyright (c) 1992-2010 VT MAK
 ** All rights reserved.
 *********************************************************************/

#ifndef smFixedMessageQueue_H_
#define smFixedMessageQueue_H_


 
#include <portal/messageQueue.h>
#include <vlutil/vlShmPool.h>
#include <vlutil/vlNetTypes.h>
#include <vector>

#include <pthread.h>

namespace MAKVrExchange {

//
//
class DT_DLL_LIBPORTAL DtSubscribableMessageQueue: public DtMessageQueue
{

public:

   DtSubscribableMessageQueue(const DtString &name, DtU8 subscriberId);
   virtual ~DtSubscribableMessageQueue();

private:

   DtSubscribableMessageQueue(const DtSubscribableMessageQueue &other);

   DtSubscribableMessageQueue& operator=(const DtSubscribableMessageQueue &other);

public:

   virtual bool isEmpty();

   virtual bool isSubscribed(DtU8 subscriberId) const;

   virtual DtU32 subscriberDrops(DtU8 subscriberId) const;

   //
   virtual void sendMessage(const void *message, unsigned int size);

   //
   virtual void* message(unsigned int *outSize = 0);


   //
   virtual void unsubscribe(DtU8 id);

   //
   void subscribe(DtU8 id);

   //
   virtual void resetQueue(DtU32 numberOfBuckets, DtU32 payloadSize);

   //
   //This should only be called after the queue is created, and only by 
   virtual void initLocks();

   virtual void clearLocks();

   virtual std::ostream &printDataToStream(std::ostream &str) const;

public:

   struct DtQueueHeader;
   struct DtBucketHeader;

   //
   static unsigned int poolSize(DtU32 numberOfBuckets, DtU32 payloadSize);

   virtual DtString name() const;


   DtQueueHeader *queueHeader() const;

   DtU32 numberOfBuckets() const;

   DtU32 payloadSize() const;

   virtual char* payloadInBucket(unsigned int bucketNum) const;

   virtual DtBucketHeader* bucketHeader(unsigned int bucketNum) const;

   static const unsigned int theMaxSubscribers = 255;

protected:

   virtual void lockForRead();
   virtual void unlockForRead();

   virtual void lockForWrite();

   virtual void unlockForWrite();

protected:

   DtSharedMemoryPoolClient *myShmPool;
   DtString myShmName;
   DtU8     mySubscriberId;
   std::vector<char> myReadBuffer;



public:

   struct DtBucketHeader
   {
      DtNetU8  senderId; 
      DtNetU32 sizeInBytes;
   };

   struct DtQueueHeader
   {
      DtNetU32 payloadSize;
      DtNetU32 numberOfBuckets;
      DtNetU32 front[theMaxSubscribers];
      DtNetU32 subscribed[theMaxSubscribers];
      DtNetU32 subscriberDrops[theMaxSubscribers];
      DtNetU32 back;
      pthread_rwlock_t rwlock;

      //The following are used strictly for performance information
      DtNetU32 lastXPtr;
      DtNetU32 lastXSize;
      DtNetU32 currentTrackableSize;
      DtNetU32 maxTrackableSize;
   };
};
}

#endif //end smFixedMessageQueue_H_