Implements a generic shared memory message queue for multiple subscribers. More...

Inheritance diagram for DtSmSubscribableMessageQueue__rti__:

Collaboration diagram for DtSmSubscribableMessageQueue__rti__:

Classes
struct	DtBucketHeader
	DtBucketHeader is a structure found at the front of each message to keep track of the number of bytes used, fragmented message indications, the subscriber that enqueued the message, and the sender's sequence number. More...
struct	DtQueueHeader
	DtQueueHeader exists at the front of the message queue and contains useful information about the state of the queue. More...
Public Member Functions
	DtSmSubscribableMessageQueue__rti__ (const MAKRti::DtString &name)
	Creates an instance of a Shared Memory Queue with the given name IMPORTANT: memory with the given name must have already been alocated somewhere, This class will only attach to it.
virtual	~DtSmSubscribableMessageQueue__rti__ ()
virtual int	subscribe (MAKRti::DtU32 subscriberId=DtSM_INVALID_SUBSCRIBER_ID)
	Subscribe to the queue and set up subscriber-specific fields in the queue header.
virtual void	unsubscribe (MAKRti::DtU32 id)
	Unsubscribes subscriber id from the queue.
virtual bool	isSubscribed (MAKRti::DtU32 subscriberId) const
	Test if subscriberId is subscribed.
virtual MAKRti::DtU32	subscriberId () const
	Return the assigned subscriber ID.
virtual int	subscriberCount () const
	Return the count of subscribers to the queue.
virtual bool	isEmpty ()
	Returns true if the subscriber's queue has no unread messages.
virtual bool	isEmpty (int *nextSize, MAKRti::DtU32 flagsMask=DtSMQ_FETCH_DEFAULT)
virtual void	setDefaultReadBehavior (MAKRti::DtU32 flagsMask)
	Set the default behavior for detecting messages in the queue to skip or return messages with certain characteristics.
virtual bool	sendMessage (const void *message, unsigned int size)
	Sends a packet at address message of size bytes to the queue.
virtual void *	message (unsigned int *outSize=0)
	Returns a pointer to the message copied from shared memory or null if there are no messages to read.
virtual void *	message (void recvBuffer, unsigned int outSize)
virtual void	resetQueue (MAKRti::DtU32 numberOfBuckets, MAKRti::DtU32 payloadSize)
	Emptys the queue.
virtual bool	sm_shutdown (bool ruthless=false)
	Shutdown the queue. Does not free shared memory or other resources.
virtual volatile MAKRti::DtU32	queueState () const
	returns current state indicating whether shutdown in progress
virtual void	initSyncVars ()
	Initializes Shared Memory locks.
virtual void	clearSyncVars ()
	This clears all synchronization variables and should only be called by the queue manager.
virtual std::ostream &	printDataToStream (std::ostream &str) const
	prints the state of the queue to the given stream
Static Public Member Functions
static unsigned int	poolSize (MAKRti::DtU32 numberOfBuckets, MAKRti::DtU32 payloadSize)
	Used to compute the size in bytes, required for a DtSmSubscribableMessageQueue instance using numberOfBuckets with a payload size of payloadSize.
Protected Member Functions
virtual int	assignSubscriberId ()
	Requests next available subscriber ID.
virtual unsigned int	maxSubscribers () const
	Report configured maximum subscriber count.
volatile DtQueueHeader *	queueHeader () const
	returns a pointer to the queue header.
virtual MAKRti::DtU32	queueHeaderSize () const
	report the size, in bytes, of the queue header
virtual MAKRti::DtU32	numberOfBuckets () const
	Returns the number of buckets in this queue.
virtual MAKRti::DtU32	payloadSize () const
	Returns the number of bytes available in an empty bucket payload.
virtual volatile MAKRti::DtU32	queueFront (unsigned int sub) const
	Returns the index of the specified subscribers queue front, which.
virtual volatile MAKRti::DtU32	queueBack () const
	Returns the back of this queue, which is the last bucket written to.
virtual volatile MAKRti::DtU32	queueBackstop () const
	Returns the backstop of this queue, which is the last bucket all subscribers have read.
virtual MAKRti::DtU32	updateFront (unsigned int sub, unsigned int val)
	Sets a new value for the specified subscriber's queue front.
virtual MAKRti::DtU32	updateBack (unsigned int val)
	Sets a new value for the queue's back.
virtual MAKRti::DtU32	updateBackstop (unsigned int val)
	Sets a new value for the queue's backstop.
virtual void	setSubscribed (unsigned int sub, unsigned int val)
	Set or clear the subscriber's subscribed bit in the queue header.
virtual int	setSubscriberCount (MAKRti::DtU32 val)
	Set the number of current subscribers.
virtual int	incrementSubscriberCount ()
	Increment the number of subscribers by one.
virtual int	decrementSubscriberCount ()
	Decrement the number of subscribers by one.
virtual void	setState (MAKRti::DtU32 val)
	Set the queue state.
virtual volatile char *	payloadInBucket (unsigned int bucketNum) const
	returns a pointer to the payload in the bucket.
virtual volatile void *	bucketHeader (unsigned int bucketNum) const
	returns a pointer to the message header at index
virtual MAKRti::DtU32	bucketHeaderSize () const
	report the size, in bytes, of the bucket header
virtual MAKRti::DtU32	bucketSenderId (unsigned int bucketNum) const
	return the bucket header field values
virtual MAKRti::DtU32	bucketMsgSn (unsigned int bucketNum) const
virtual MAKRti::DtU32	bucketOutstandingReadCount (unsigned int bucketNum) const
virtual MAKRti::DtU32	bucketPayloadLen (unsigned int bucketNum) const
virtual bool	bucketIsFirst (unsigned int bucketNum) const
virtual bool	bucketIsLast (unsigned int bucketNum) const
virtual void	setSenderId (unsigned int bucketNum, MAKRti::DtU32 sid)
	change the value of bucket header fields
virtual void	setMsgSn (unsigned int bucketNum, MAKRti::DtU32 seqnum)
virtual void	setOutstandingReadCount (unsigned int bucketNum, MAKRti::DtU32 val)
virtual void	decrementOutstandingReadCount (unsigned int bucketNum)
virtual void	setPayloadLen (unsigned int bucketNum, MAKRti::DtU32 len)
virtual void	setFirst (unsigned int bucketNum, MAKRti::DtU32 val)
virtual void	setLast (unsigned int bucketNum, MAKRti::DtU32 val)
virtual void	lockForRead ()
	Locks the queue for a read.
virtual void	unlockForRead ()
	Unlocks the queue for a read.
virtual void	lockForWrite ()
	Locks the queue for a write.
virtual void	unlockForWrite ()
	Unlocks the queue for writes.
virtual bool	stalledSubscribers (unsigned int bucketNum, unsigned int numStalled)
	Record information about stalled subscriber.
virtual bool	checkForStalls ()
	Determine if there are any subscribers stalled on the bucket immediately following the 'backstop' bucket.
virtual pthread_rwlock_t *	rdWrLock ()
	For non-Win32 implementations the rwlock structure in the queue header must be accessed via inheritable methods in order for derived classes to be able to define a different queue header structure.
virtual pthread_mutex_t *	orcMutex ()
Protected Attributes
DtSharedMemoryPoolClient__rti__ *	myShmPool
	!WIN32
MAKRti::DtString	myShmName
MAKRti::DtU32	mySubscriberId
MAKRti::DtU32	myMessagesSentCnt
DtIpcRdWrLock__rti__ *	myRdWrLock
DtIpcMutex__rti__ *	myOrcMutex
std::vector< char >	myReadBuffer
MAKRti::DtU32	myDefaultReadMask
unsigned int	theMaxSubscribers
	The maximum number of subscribers allowed in this queue.
Private Member Functions
	DtSmSubscribableMessageQueue__rti__ (const DtSmSubscribableMessageQueue__rti__ &other)
	Copy Ctor Not Implemented -- Class Not Duplicable.
DtSmSubscribableMessageQueue__rti__ &	operator= (const DtSmSubscribableMessageQueue__rti__ &other)
	Assignment Operator Not Implimented -- Class Not Duplicable.

Detailed Description

Implements a generic shared memory message queue for multiple subscribers.

This is a fixed length queue comprised of fixed length message containers. It is implemented as a circular list that will not fill up. Any subscriber which has a queue full of unread messages will be unsubscribed. Subscribers are given a 32 bit ID and register their interest in messages merely by creating an instance of this class. A subscriber can unsubscribe themselves at any time. Before an instance of this class can be created there must already be a shared memory pool of the correct size open. The correct size is computed using DtSubscribableMessageQueue::poolSize(). Before a queue can be used one subscriber (and only one subscriber) must call queue->reset(...); queue->initSyncVars();

Constructor & Destructor Documentation

DtSmSubscribableMessageQueue__rti__::DtSmSubscribableMessageQueue__rti__ ( const MAKRti::DtString & name )

Creates an instance of a Shared Memory Queue with the given name IMPORTANT: memory with the given name must have already been alocated somewhere, This class will only attach to it.

The memory must be at least as big as ::poolSize(), this queue can't check.

DtSmSubscribableMessageQueue__rti__::~DtSmSubscribableMessageQueue__rti__ ( ) [virtual]

References DtSM_INVALID_SUBSCRIBER_ID, myRdWrLock, myShmPool, mySubscriberId, and unsubscribe().

DtSmSubscribableMessageQueue__rti__::DtSmSubscribableMessageQueue__rti__ ( const DtSmSubscribableMessageQueue__rti__ & other ) [private]

Copy Ctor Not Implemented -- Class Not Duplicable.

Member Function Documentation

int DtSmSubscribableMessageQueue__rti__::assignSubscriberId ( ) [protected, virtual]

Requests next available subscriber ID.

References DtSM_INVALID_SUBSCRIBER_ID, incrementSubscriberCount(), isSubscribed(), lockForWrite(), maxSubscribers(), myShmName, setSubscribed(), subscriberId(), and unlockForWrite().

Referenced by subscribe().

volatile void * DtSmSubscribableMessageQueue__rti__::bucketHeader ( unsigned int bucketNum ) const [protected, virtual]

returns a pointer to the message header at index

References DtSharedMemoryPoolClient__rti__::address(), bucketHeaderSize(), myShmPool, NULL, numberOfBuckets(), payloadSize(), and queueHeaderSize().

Referenced by bucketIsFirst(), bucketIsLast(), bucketMsgSn(), bucketOutstandingReadCount(), bucketPayloadLen(), bucketSenderId(), decrementOutstandingReadCount(), payloadInBucket(), printDataToStream(), resetQueue(), setFirst(), setLast(), setMsgSn(), setOutstandingReadCount(), setPayloadLen(), and setSenderId().

DtU32 DtSmSubscribableMessageQueue__rti__::bucketHeaderSize ( ) const [protected, virtual]

report the size, in bytes, of the bucket header

Referenced by bucketHeader(), and payloadInBucket().

bool DtSmSubscribableMessageQueue__rti__::bucketIsFirst ( unsigned int bucketNum ) const [protected, virtual]

References bucketHeader().

Referenced by message().

bool DtSmSubscribableMessageQueue__rti__::bucketIsLast ( unsigned int bucketNum ) const [protected, virtual]

References bucketHeader().

Referenced by message().

DtU32 DtSmSubscribableMessageQueue__rti__::bucketMsgSn ( unsigned int bucketNum ) const [protected, virtual]

References bucketHeader().

Referenced by message().

DtU32 DtSmSubscribableMessageQueue__rti__::bucketOutstandingReadCount ( unsigned int bucketNum ) const [protected, virtual]

References bucketHeader().

Referenced by checkForStalls().

DtU32 DtSmSubscribableMessageQueue__rti__::bucketPayloadLen ( unsigned int bucketNum ) const [protected, virtual]

References bucketHeader().

Referenced by message().

DtU32 DtSmSubscribableMessageQueue__rti__::bucketSenderId ( unsigned int bucketNum ) const [protected, virtual]

return the bucket header field values

References bucketHeader().

Referenced by message().

bool DtSmSubscribableMessageQueue__rti__::checkForStalls ( ) [protected, virtual]

Determine if there are any subscribers stalled on the bucket immediately following the 'backstop' bucket.

References bucketOutstandingReadCount(), numberOfBuckets(), queueBackstop(), and stalledSubscribers().

Referenced by sendMessage().

void DtSmSubscribableMessageQueue__rti__::clearSyncVars ( ) [virtual]

This clears all synchronization variables and should only be called by the queue manager.

References DtIpcRdWrLock__rti__::destroy(), myOrcMutex, myRdWrLock, myShmName, and mySubscriberId.

Referenced by sm_shutdown().

void DtSmSubscribableMessageQueue__rti__::decrementOutstandingReadCount ( unsigned int bucketNum ) [protected, virtual]

References bucketHeader(), DtIpcMutex__rti__::lock(), myOrcMutex, DtIpcMutex__rti__::unlock(), and updateBackstop().

Referenced by message(), and unsubscribe().

int DtSmSubscribableMessageQueue__rti__::decrementSubscriberCount ( ) [protected, virtual]

Decrement the number of subscribers by one.

References DtSmSubscribableMessageQueue__rti__::DtQueueHeader::numberOfSubscribers, queueHeader(), and subscriberCount().

Referenced by unsubscribe().

int DtSmSubscribableMessageQueue__rti__::incrementSubscriberCount ( ) [protected, virtual]

Increment the number of subscribers by one.

References DtSmSubscribableMessageQueue__rti__::DtQueueHeader::numberOfSubscribers, queueHeader(), subscriberCount(), and theMaxSubscribers.

Referenced by assignSubscriberId(), resetQueue(), and subscribe().

void DtSmSubscribableMessageQueue__rti__::initSyncVars ( ) [virtual]

Initializes Shared Memory locks.

This should only be called after the queue is created, and only by one party, the queue manager.

References DtIpcRdWrLock__rti__::init(), DtIpcMutex__rti__::init(), myOrcMutex, myRdWrLock, myShmName, and mySubscriberId.

bool DtSmSubscribableMessageQueue__rti__::isEmpty ( ) [virtual]

Returns true if the subscriber's queue has no unread messages.

Returns true if the queue has messages no messages in it, false if there is a message in the queue. If nextSize is not NULL the size of the next message is stored in that location. The flagsMask argument can control which messages the subscriber will detect in the queue (overrides default read behavior)

Implements DtMessageQueue__rti__.

References NULL.

Referenced by message(), and unsubscribe().

virtual bool DtSmSubscribableMessageQueue__rti__::isEmpty	(	int *	nextSize,
		MAKRti::DtU32	flagsMask = `DtSMQ_FETCH_DEFAULT`
	)		`[virtual]`

bool DtSmSubscribableMessageQueue__rti__::isSubscribed ( MAKRti::DtU32 subscriberId ) const [virtual]

Test if subscriberId is subscribed.

References queueHeader(), DtSmSubscribableMessageQueue__rti__::DtQueueHeader::subscribed, and subscriberId().

Referenced by assignSubscriberId(), message(), printDataToStream(), sendMessage(), sm_shutdown(), and subscribe().

void DtSmSubscribableMessageQueue__rti__::lockForRead ( ) [protected, virtual]

Locks the queue for a read.

References myRdWrLock, and DtRdWrLock__rti__::rdLock().

Referenced by message().

void DtSmSubscribableMessageQueue__rti__::lockForWrite ( ) [protected, virtual]

Locks the queue for a write.

References myRdWrLock, and DtRdWrLock__rti__::wrLock().

Referenced by assignSubscriberId(), sendMessage(), sm_shutdown(), and subscribe().

unsigned int DtSmSubscribableMessageQueue__rti__::maxSubscribers ( ) const [protected, virtual]

Report configured maximum subscriber count.

References theMaxSubscribers.

Referenced by assignSubscriberId(), printDataToStream(), resetQueue(), sm_shutdown(), and stalledSubscribers().

void * DtSmSubscribableMessageQueue__rti__::message ( unsigned int * outSize = 0 ) [virtual]

Returns a pointer to the message copied from shared memory or null if there are no messages to read.

If recvBuffer pointer is supplied the message will be copied to that location, rather than the intermediate local buffer. If outsize is passed then the size of the message is returned.

Implements DtMessageQueue__rti__.

References NULL.

void * DtSmSubscribableMessageQueue__rti__::message	(	void *	recvBuffer,
		unsigned int *	outSize
	)		`[virtual]`

References bucketIsFirst(), bucketIsLast(), bucketMsgSn(), bucketPayloadLen(), bucketSenderId(), decrementOutstandingReadCount(), isEmpty(), isSubscribed(), lockForRead(), myReadBuffer, myShmName, mySubscriberId, numberOfBuckets(), payloadInBucket(), payloadSize(), queueFront(), unlockForRead(), and updateFront().

DtU32 DtSmSubscribableMessageQueue__rti__::numberOfBuckets ( ) const [protected, virtual]

Returns the number of buckets in this queue.

References DtSmSubscribableMessageQueue__rti__::DtQueueHeader::numberOfBuckets, and queueHeader().

Referenced by bucketHeader(), checkForStalls(), message(), payloadInBucket(), printDataToStream(), resetQueue(), sendMessage(), unsubscribe(), updateBack(), updateBackstop(), and updateFront().

DtSmSubscribableMessageQueue__rti__& DtSmSubscribableMessageQueue__rti__::operator= ( const DtSmSubscribableMessageQueue__rti__ & other ) [private]

Assignment Operator Not Implimented -- Class Not Duplicable.

pthread_mutex_t * DtSmSubscribableMessageQueue__rti__::orcMutex ( ) [protected, virtual]

References queueHeader().

Referenced by subscribe().

volatile char * DtSmSubscribableMessageQueue__rti__::payloadInBucket ( unsigned int bucketNum ) const [protected, virtual]

returns a pointer to the payload in the bucket.

Where a bucket is composed of the bucketHeader and the bucketPayload.

References bucketHeader(), bucketHeaderSize(), NULL, and numberOfBuckets().

Referenced by message(), and sendMessage().

DtU32 DtSmSubscribableMessageQueue__rti__::payloadSize ( ) const [protected, virtual]

Returns the number of bytes available in an empty bucket payload.

References DtSmSubscribableMessageQueue__rti__::DtQueueHeader::payloadSize, and queueHeader().

Referenced by bucketHeader(), message(), resetQueue(), and sendMessage().

unsigned int DtSmSubscribableMessageQueue__rti__::poolSize	(	MAKRti::DtU32	numberOfBuckets,
		MAKRti::DtU32	payloadSize
	)		`[static]`

Used to compute the size in bytes, required for a DtSmSubscribableMessageQueue instance using numberOfBuckets with a payload size of payloadSize.

Each bucket is the size of the payload + the size of a bucketHeader

std::ostream & DtSmSubscribableMessageQueue__rti__::printDataToStream ( std::ostream & str ) const [virtual]

prints the state of the queue to the given stream

Implements DtMessageQueue__rti__.

References bucketHeader(), DtSmSubscribableMessageQueue__rti__::DtBucketHeader::first, isSubscribed(), DtSmSubscribableMessageQueue__rti__::DtBucketHeader::last, maxSubscribers(), DtSmSubscribableMessageQueue__rti__::DtBucketHeader::msgSn, myShmName, numberOfBuckets(), queueBack(), queueFront(), DtSmSubscribableMessageQueue__rti__::DtBucketHeader::senderId, and DtSmSubscribableMessageQueue__rti__::DtBucketHeader::sizeInBytes.

volatile DtU32 DtSmSubscribableMessageQueue__rti__::queueBack ( ) const [protected, virtual]

Returns the back of this queue, which is the last bucket written to.

References queueHeader().

Referenced by printDataToStream(), sendMessage(), subscribe(), and unsubscribe().

volatile DtU32 DtSmSubscribableMessageQueue__rti__::queueBackstop ( ) const [protected, virtual]

Returns the backstop of this queue, which is the last bucket all subscribers have read.

References queueHeader().

Referenced by checkForStalls(), and sendMessage().

volatile DtU32 DtSmSubscribableMessageQueue__rti__::queueFront ( unsigned int sub ) const [protected, virtual]

Returns the index of the specified subscribers queue front, which.

is the last bucket the subscriber read from.

References queueHeader().

Referenced by message(), printDataToStream(), and unsubscribe().

volatile DtSmSubscribableMessageQueue__rti__::DtQueueHeader * DtSmSubscribableMessageQueue__rti__::queueHeader ( ) const [protected]

returns a pointer to the queue header.

References DtSharedMemoryPoolClient__rti__::address(), and myShmPool.

Referenced by decrementSubscriberCount(), incrementSubscriberCount(), isSubscribed(), numberOfBuckets(), orcMutex(), payloadSize(), queueBack(), queueBackstop(), queueFront(), queueState(), rdWrLock(), resetQueue(), setState(), setSubscribed(), setSubscriberCount(), stalledSubscribers(), subscriberCount(), updateBack(), updateBackstop(), and updateFront().

DtU32 DtSmSubscribableMessageQueue__rti__::queueHeaderSize ( ) const [protected, virtual]

report the size, in bytes, of the queue header

Referenced by bucketHeader().

volatile DtU32 DtSmSubscribableMessageQueue__rti__::queueState ( ) const [virtual]

returns current state indicating whether shutdown in progress

References queueHeader(), and DtSmSubscribableMessageQueue__rti__::DtQueueHeader::state.

Referenced by subscribe().

pthread_rwlock_t * DtSmSubscribableMessageQueue__rti__::rdWrLock ( ) [protected, virtual]

For non-Win32 implementations the rwlock structure in the queue header must be accessed via inheritable methods in order for derived classes to be able to define a different queue header structure.

References queueHeader().

Referenced by subscribe().

void DtSmSubscribableMessageQueue__rti__::resetQueue	(	MAKRti::DtU32	numberOfBuckets,
		MAKRti::DtU32	payloadSize
	)		`[virtual]`

Emptys the queue.

This should only be called by the queue manager, as all unread messages are lost and all subscribers are unsubscribed.

bool DtSmSubscribableMessageQueue__rti__::sendMessage	(	const void *	message,
		unsigned int	size
	)		`[virtual]`

Sends a packet at address message of size bytes to the queue.

If the queue is full this message will knock the oldest message off the queue.

Implements DtMessageQueue__rti__.

References checkForStalls(), isSubscribed(), lockForWrite(), myMessagesSentCnt, myShmName, mySubscriberId, NULL, numberOfBuckets(), payloadInBucket(), payloadSize(), queueBack(), queueBackstop(), setFirst(), setLast(), setMsgSn(), setPayloadLen(), setSenderId(), unlockForWrite(), and updateBack().

void DtSmSubscribableMessageQueue__rti__::setDefaultReadBehavior ( MAKRti::DtU32 flagsMask ) [virtual]

Set the default behavior for detecting messages in the queue to skip or return messages with certain characteristics.

References myDefaultReadMask.

void DtSmSubscribableMessageQueue__rti__::setFirst	(	unsigned int	bucketNum,
		MAKRti::DtU32	val
	)		`[protected, virtual]`

References bucketHeader().

Referenced by sendMessage().

void DtSmSubscribableMessageQueue__rti__::setLast	(	unsigned int	bucketNum,
		MAKRti::DtU32	val
	)		`[protected, virtual]`

References bucketHeader().

Referenced by sendMessage().

void DtSmSubscribableMessageQueue__rti__::setMsgSn	(	unsigned int	bucketNum,
		MAKRti::DtU32	seqnum
	)		`[protected, virtual]`

References bucketHeader().

Referenced by sendMessage().

void DtSmSubscribableMessageQueue__rti__::setOutstandingReadCount	(	unsigned int	bucketNum,
		MAKRti::DtU32	val
	)		`[protected, virtual]`

References bucketHeader().

void DtSmSubscribableMessageQueue__rti__::setPayloadLen	(	unsigned int	bucketNum,
		MAKRti::DtU32	len
	)		`[protected, virtual]`

References bucketHeader().

Referenced by sendMessage().

void DtSmSubscribableMessageQueue__rti__::setSenderId	(	unsigned int	bucketNum,
		MAKRti::DtU32	sid
	)		`[protected, virtual]`

change the value of bucket header fields

References bucketHeader().

Referenced by sendMessage().

void DtSmSubscribableMessageQueue__rti__::setState ( MAKRti::DtU32 val ) [protected, virtual]

Set the queue state.

References queueHeader(), and DtSmSubscribableMessageQueue__rti__::DtQueueHeader::state.

Referenced by resetQueue(), and sm_shutdown().

void DtSmSubscribableMessageQueue__rti__::setSubscribed	(	unsigned int	sub,
		unsigned int	val
	)		`[protected, virtual]`

Set or clear the subscriber's subscribed bit in the queue header.

References mySubscriberId, queueHeader(), and DtSmSubscribableMessageQueue__rti__::DtQueueHeader::subscribed.

Referenced by assignSubscriberId(), resetQueue(), sm_shutdown(), subscribe(), and unsubscribe().

int DtSmSubscribableMessageQueue__rti__::setSubscriberCount ( MAKRti::DtU32 val ) [protected, virtual]

Set the number of current subscribers.

References DtSmSubscribableMessageQueue__rti__::DtQueueHeader::numberOfSubscribers, queueHeader(), subscriberCount(), and theMaxSubscribers.

bool DtSmSubscribableMessageQueue__rti__::sm_shutdown ( bool ruthless = false ) [virtual]

Shutdown the queue. Does not free shared memory or other resources.

References clearSyncVars(), DtSm_QUEUE_STATE_SHUTDOWN, isSubscribed(), lockForWrite(), maxSubscribers(), mySubscriberId, setState(), setSubscribed(), and unlockForWrite().

bool DtSmSubscribableMessageQueue__rti__::stalledSubscribers	(	unsigned int	bucketNum,
		unsigned int	numStalled
	)		`[protected, virtual]`

Record information about stalled subscriber.

Unsubscribe stalled subscribers if necessary

References maxSubscribers(), myShmName, mySubscriberId, queueHeader(), and unsubscribe().

Referenced by checkForStalls().

int DtSmSubscribableMessageQueue__rti__::subscribe ( MAKRti::DtU32 subscriberId = DtSM_INVALID_SUBSCRIBER_ID ) [virtual]

Subscribe to the queue and set up subscriber-specific fields in the queue header.

Return 0 on failure, 1 on success

References assignSubscriberId(), DtSM_INVALID_SUBSCRIBER_ID, DtSm_QUEUE_STATE_INITIALIZED, incrementSubscriberCount(), isSubscribed(), lockForWrite(), myOrcMutex, myRdWrLock, myShmName, mySubscriberId, orcMutex(), queueBack(), queueState(), rdWrLock(), setSubscribed(), unlockForWrite(), and updateFront().

int DtSmSubscribableMessageQueue__rti__::subscriberCount ( ) const [virtual]

Return the count of subscribers to the queue.

References DtSmSubscribableMessageQueue__rti__::DtQueueHeader::numberOfSubscribers, and queueHeader().

Referenced by decrementSubscriberCount(), incrementSubscriberCount(), and setSubscriberCount().

DtU32 DtSmSubscribableMessageQueue__rti__::subscriberId ( ) const [virtual]

Return the assigned subscriber ID.

References mySubscriberId.

Referenced by assignSubscriberId(), and isSubscribed().

void DtSmSubscribableMessageQueue__rti__::unlockForRead ( ) [protected, virtual]

Unlocks the queue for a read.

References myRdWrLock, and DtRdWrLock__rti__::rdUnlock().

Referenced by message().

void DtSmSubscribableMessageQueue__rti__::unlockForWrite ( ) [protected, virtual]

Unlocks the queue for writes.

References myRdWrLock, and DtRdWrLock__rti__::wrUnlock().

Referenced by assignSubscriberId(), sendMessage(), sm_shutdown(), and subscribe().

void DtSmSubscribableMessageQueue__rti__::unsubscribe ( MAKRti::DtU32 id ) [virtual]

Unsubscribes subscriber id from the queue.

This is called by the DTOR and should only be called by users of this class with caution. Unsubscribing will clear subscriber ID. (except the queue manager's ID, which is always reserved as 0) The queue manager may unsubscribe other subscribers without clearing that subscriber objects subscriber ID, so should be done only in extreme cases. Once unsubscribed, use subscribe() to get a new subscriber ID.

References decrementOutstandingReadCount(), decrementSubscriberCount(), DtSM_INVALID_SUBSCRIBER_ID, DtSMQ_FETCH_ALL, isEmpty(), myShmName, mySubscriberId, NULL, numberOfBuckets(), queueBack(), queueFront(), and setSubscribed().